如何使用 asyncio 在单独的线程上通知 RxPY 观察者？答案

【问题标题】：How can I notify RxPY observers on separate threads using asyncio?如何使用 asyncio 在单独的线程上通知 RxPY 观察者？
【发布时间】：2016-09-04 15:48:23
【问题描述】：

（注意：这个问题的背景很冗长，但是底部有一个SSCCE可以跳过）

背景

我正在尝试开发基于 Python 的 CLI 来与 Web 服务交互。在我的代码库中，我有一个 CommunicationService 类来处理与 Web 服务的所有直接通信。它公开了一个 received_response 属性，该属性返回一个 Observable（来自 RxPY），其他对象可以订阅该属性，以便在从 Web 服务收到响应时得到通知。

我的 CLI 逻辑基于 click 库，其中我的子命令之一实现如下：

async def enabled(self, request: str, response_handler: Callable[[str], Tuple[bool, str]]) -> None:
    self._generate_request(request)
    if response_handler is None:
        return None

    while True:
        response = await self.on_response
        success, value = response_handler(response)
        print(success, value)
        if success:
            return value

这里发生的事情（在response_handler 不是None 的情况下）是子命令表现为等待来自Web 服务（self.on_response == CommunicationService.received_response）的响应并从第一个响应返回一些处理后的值它可以处理。

我正在尝试通过创建完全模拟 CommunicationService 的测试用例来测试我的 CLI 的行为；创建了一个假的Subject（可以充当Observable）并模拟CommunicationService.received_response 以返回它。作为测试的一部分，主题的 on_next 方法被调用以将模拟 Web 服务响应传回生产代码：

@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
    context.mock_received_response_subject.on_next(context.text)

我使用一个 click 'result callback' 函数，该函数在 CLI 调用结束时被调用并阻塞，直到协程（子命令）完成：

@cli.resultcallback()
def _handle_command_task(task: Coroutine, **_) -> None:
    if task:
        loop = asyncio.get_event_loop()
        result = loop.run_until_complete(task)
        loop.close()
        print('RESULT:', result)

问题

在测试开始时，我运行CliRunner.invoke 来启动整个shebang。问题是这是一个阻塞调用，并且会阻塞线程，直到 CLI 完成并返回结果，如果我需要我的测试线程继续运行以便它可以同时产生模拟 Web 服务响应，这将没有帮助。

我想我需要做的是使用ThreadPoolExecutor 在新线程上运行CliRunner.invoke。这允许测试逻辑在原始线程上继续并执行上面发布的@when 步骤。但是，用mock_received_response_subject.on_next 发布的通知似乎不会触发在子命令中继续执行。

我相信解决方案将涉及使用 RxPY 的 AsyncIOScheduler，但我发现这方面的文档有点稀疏而且没有帮助。

SSCCE

下面的 sn-p 抓住了我希望是问题的本质。如果可以修改它以使其工作，我应该能够将相同的解决方案应用于我的实际代码以使其按我想要的方式运行。

import asyncio
import logging
import sys
import time

import click
from click.testing import CliRunner
from rx.subjects import Subject

web_response_subject = Subject()
web_response_observable = web_response_subject.as_observable()

thread_loop = asyncio.new_event_loop()


@click.group()
def cli():
    asyncio.set_event_loop(thread_loop)


@cli.resultcallback()
def result_handler(task, **_):
    loop = asyncio.get_event_loop()
    result = loop.run_until_complete(task) # Should block until subject publishes value
    loop.close()

    print(result)


@cli.command()
async def get_web_response():
    return await web_response_observable


def test():
    runner = CliRunner()
    future = thread_loop.run_in_executor(None, runner.invoke, cli, ['get_web_response'])
    time.sleep(1)
    web_response_subject.on_next('foo') # Simulate reception of web response.
    time.sleep(1)
    result = future.result()
    print(result.output)

logging.basicConfig(
    level=logging.DEBUG,
    format='%(threadName)10s %(name)18s: %(message)s',
    stream=sys.stderr,
)

test()

当前行为

程序在运行时挂起，阻塞在result = loop.run_until_complete(task)。

接受标准

程序终止并在stdout 上打印foo。

更新 1

在 Vincent 的帮助下，我对代码进行了一些更改。

Relay.enabled（等待来自 Web 服务的响应以处理它们的子命令）现在实现如下：

async def enabled(self, request: str, response_handler: Callable[[str], Tuple[bool, str]]) -> None:
    self._generate_request(request)

    if response_handler is None:
        return None

    return await self.on_response \
        .select(response_handler) \
        .where(lambda result, i: result[0]) \
        .select(lambda result, index: result[1]) \
        .first()

我不太确定await 与RxPY observables 的表现如何——它们会在每个生成的元素上将执行返回给调用者，还是仅在observable 完成（或出错？）时返回执行。我现在知道是后者，老实说，这感觉像是更自然的选择，并且让我可以让这个函数的实现感觉更加优雅和反应灵敏。

我还修改了生成模拟 Web 服务响应的测试步骤：

@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
    loop = asyncio.get_event_loop()
    loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, context.text)

不幸的是，这将无法正常工作，因为 CLI 正在其自己的线程中被调用...

@when('the CLI is run with "{arguments}"')
def step_impl(context, arguments):
    loop = asyncio.get_event_loop()
    if 'async.cli' in context.tags:
        context.async_result = loop.run_in_executor(None, context.cli_runner.invoke, testcube.cli, arguments.split())
    else:
        ...

CLI 在调用时会创建自己的线程私有事件循环...

def cli(context, hostname, port):
    _initialize_logging(context.meta['click_log.core.logger']['level'])

    # Create a new event loop for processing commands asynchronously on.
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    ...

我认为我需要一种方法来允许我的测试步骤在新线程上调用 CLI然后获取它正在使用的事件循环：

@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
    loop = _get_cli_event_loop() # Needs to be implemented.
    loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, context.text)

更新 2

似乎没有一种简单的方法来获取特定线程为其自身创建和使用的事件循环，因此我接受了 Victor 的建议并模拟了 asyncio.new_event_loop 以返回我的测试代码创建的事件循环并商店：

def _apply_mock_event_loop_patch(context):
    # Close any already-existing exit stacks.
    if hasattr(context, 'mock_event_loop_exit_stack'):
        context.mock_event_loop_exit_stack.close()

    context.test_loop = asyncio.new_event_loop()
    print(context.test_loop)
    context.mock_event_loop_exit_stack = ExitStack()
    context.mock_event_loop_exit_stack.enter_context(
        patch.object(asyncio, 'new_event_loop', spec=True, return_value=context.test_loop))

我将“收到的模拟网络响应”测试步骤更改为执行以下操作：

@when('the communications service receives a response from TestCube Web Service')
def step_impl(context):
    loop = context.test_loop
    loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, context.text)

好消息是，当执行此步骤时，我实际上正在触发 Relay.enabled 协程！

现在唯一的问题是最后的测试步骤，我等待在自己的线程中执行 CLI 并验证 CLI 是否在 stdout 上发送此消息：

@then('the CLI should print "{output}"')
def step_impl(context, output):
    if 'async.cli' in context.tags:
        loop = asyncio.get_event_loop() # main loop, not test loop
        result = loop.run_until_complete(context.async_result)
    else:
        result = context.result
    assert_that(result.output, equal_to(output))

我试过玩这个，但我似乎无法让context.async_result（它存储来自loop.run_in_executor 的未来）很好地过渡到done 并返回结果。使用当前的实现，我在第一次测试 (1.1) 时遇到错误，在第二次 (1.2) 时出现无限期挂起：

 @mock.comms @async.cli @wip
  Scenario Outline: Querying relay enable state -- @1.1                           # testcube/tests/features/relay.feature:45
    When the user queries the enable state of relay 0                             # testcube/tests/features/steps/relay.py:17 0.003s
    Then the CLI should query the web service about the enable state of relay 0   # testcube/tests/features/steps/relay.py:48 0.000s
    When the communications service receives a response from TestCube Web Service # testcube/tests/features/steps/core.py:58 0.000s
      """
      {'module':'relays','path':'relays[0].enabled','data':[True]}'
      """
    Then the CLI should print "True"                                              # testcube/tests/features/steps/core.py:94 0.003s
      Traceback (most recent call last):
        File "/Users/davidfallah/testcube_env/lib/python3.5/site-packages/behave/model.py", line 1456, in run
          match.run(runner.context)
        File "/Users/davidfallah/testcube_env/lib/python3.5/site-packages/behave/model.py", line 1903, in run
          self.func(context, *args, **kwargs)
        File "testcube/tests/features/steps/core.py", line 99, in step_impl
          result = loop.run_until_complete(context.async_result)
        File "/usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/base_events.py", line 387, in run_until_complete
          return future.result()
        File "/usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py", line 274, in result
          raise self._exception
        File "/usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/concurrent/futures/thread.py", line 55, in run
          result = self.fn(*self.args, **self.kwargs)
        File "/Users/davidfallah/testcube_env/lib/python3.5/site-packages/click/testing.py", line 299, in invoke
          output = out.getvalue()
      ValueError: I/O operation on closed file.

      Captured stdout:
      RECEIVED WEB RESPONSE: {'module':'relays','path':'relays[0].enabled','data':[True]}'
      <Future pending cb=[_chain_future.<locals>._call_check_cancel() at /usr/local/Cellar/python3/3.5.2_1/Frameworks/Python.framework/Versions/3.5/lib/python3.5/asyncio/futures.py:431]>

  @mock.comms @async.cli @wip
  Scenario Outline: Querying relay enable state -- @1.2                           # testcube/tests/features/relay.feature:46
    When the user queries the enable state of relay 1                             # testcube/tests/features/steps/relay.py:17 0.005s
    Then the CLI should query the web service about the enable state of relay 1   # testcube/tests/features/steps/relay.py:48 0.001s
    When the communications service receives a response from TestCube Web Service # testcube/tests/features/steps/core.py:58 0.000s
      """
      {'module':'relays','path':'relays[1].enabled','data':[False]}'
      """
RECEIVED WEB RESPONSE: {'module':'relays','path':'relays[1].enabled','data':[False]}'
    Then the CLI should print "False"                                             # testcube/tests/features/steps/core.py:94

第 3 章：结局

搞砸所有这些异步多线程的东西，我太笨了。

首先，不要像这样描述场景......

When the user queries the enable state of relay <relay_id>
Then the CLI should query the web service about the enable state of relay <relay_id>
When the communications service receives a response from TestCube Web Service:
  """
  {"module":"relays","path":"relays[<relay_id>].enabled","data":[<relay_enabled>]}
  """
Then the CLI should print "<relay_enabled>"

我们是这样描述的：

Given the communications service will respond to requests:
  """
  {"module":"relays","path":"relays[<relay_id>].enabled","data":[<relay_enabled>]}
  """
When the user queries the enable state of relay <relay_id>
Then the CLI should query the web service about the enable state of relay <relay_id>
And the CLI should print "<relay_enabled>"

实施新的给定步骤：

@given('the communications service will respond to requests')
def step_impl(context):
    response = context.text

    def publish_mock_response(_):
        loop = context.test_loop
        loop.call_soon_threadsafe(context.mock_received_response_subject.on_next, response)

    # Configure the mock comms service to publish a mock response when a request is made.
    instance = context.mock_comms.return_value
    instance.send_request.on_next.side_effect = publish_mock_response

轰隆隆

2 features passed, 0 failed, 0 skipped
22 scenarios passed, 0 failed, 0 skipped
58 steps passed, 0 failed, 0 skipped, 0 undefined
Took 0m0.111s

【问题讨论】：

这个问题让我很开心 :)

标签： python multithreading python-3.x python-asyncio rx-py

【解决方案1】：

我发现你的代码有两个问题：

asyncio 不是线程安全的，除非你使用call_soon_threadsafe 或run_coroutine_threadsafe。 RxPy 不使用 Observable.to_future 中的任何对象，因此您必须在运行 asyncio 事件循环的同一线程中访问 RxPy 对象。
RxPy 在调用 on_completed 时设置未来的结果，以便等待可观察对象返回最后一个发出的对象。这意味着您必须同时调用 on_next 和 on_completed 才能让 await 返回。

这是一个工作示例：

import click
import asyncio
from rx.subjects import Subject
from click.testing import CliRunner

web_response_subject = Subject()
web_response_observable = web_response_subject.as_observable()
main_loop = asyncio.get_event_loop()

@click.group()
def cli():
    pass

@cli.resultcallback()
def result_handler(task, **_):
    future = asyncio.run_coroutine_threadsafe(task, main_loop)
    print(future.result())

@cli.command()
async def get_web_response():
    return await web_response_observable

def test():
    runner = CliRunner()
    future = main_loop.run_in_executor(
        None, runner.invoke, cli, ['get_web_response'])
    main_loop.call_later(1, web_response_subject.on_next, 'foo')
    main_loop.call_later(2, web_response_subject.on_completed)
    result = main_loop.run_until_complete(future)
    print(result.output, end='')

if __name__ == '__main__':
    test()

【讨论】：

谢谢，非常感谢您的帮助。这很好用，我想我已经取得了一些进展（我已经更新了我的帖子）。我不认为有办法检查线程内设置的事件循环？某种查询异步全局上下文的方法可能吗？我的生产代码不应该知道或关心我的测试代码，因此它将无法使用测试代码创建的事件循环（除非我可能对 unittest.mock 做了一些疯狂的事情）。
@Tagc 我不确定我是否理解您的所有限制，但我的建议是保持简单，并在您需要将任何内容注入生产代码时使用模拟（例如补丁new_event_loop返回一个测试循环）。另外，尽量不要混合事件循环：如果您确实需要在线程中运行任务，请使用ThreadPoolExecutor。
保持简单？我保持它非常简单。我正在深入 asyncio 的核心，创建一个 AbstractEventLoopPolicy 装饰器子类，它拦截了对 new_event_loop 的调用。 :) 不过，你只是嘲笑它的想法要简单得多，而且我在这方面取得了更好的成功。剩下的唯一一件事——我认为——就是让loop.run_in_executor 创建的 Future 在最后的测试步骤中整齐地完成并返回其结果。我已经更新了我的帖子。
@Tagc 我会使用concurrent.future 代替：context.async_result = context.executor.submit(context.cli_runner.invoke, testcube.cli, arguments) 然后稍后result = context.async_result.result()
@Tagc 还好你把它做成单线程的。我最近听到的一句座右铭：向程序添加线程至少会引入一个错误；）