Python asyncio 任务的产出很差

Question

我对如何使用 Python 3.4 中的 asyncio 模块感到困惑。我有一个用于搜索引擎的 searching API，并且希望每个搜索请求都是运行并行或异步的，这样我就不必等待一次搜索完成开始另一个。

这是我的高级搜索 API 以使用原始搜索结果构建一些对象。搜索引擎本身正在使用某种异步机制，因此我不会为此烦恼。

# No asyncio module used here now
class search(object):
  ...
  self.s = some_search_engine()
  ...
  def searching(self, *args, **kwargs):
    ret = {}
    # do some raw searching according to args and kwargs and build the wrapped results
    ...
    return ret

为了尝试异步请求，我编写了以下测试用例来测试我如何与 asyncio 模块进行交互。

# Here is my testing script
@asyncio.coroutine
def handle(f, *args, **kwargs):
  r = yield from f(*args, **kwargs)
  return r

s = search()
loop = asyncio.get_event_loop()
loop.run_until_complete(handle(s.searching, arg1, arg2, ...))
loop.close()

通过运行 pytest，它会return一个RuntimeError: Task got bad yield : {results from searching...}，当它命中r = yield from ...行时。

我也试过另一种方法

# same handle as above
def handle(..):
  ....
s = search()
loop = asyncio.get_event_loop()
tasks = [
        asyncio.async(handle(s.searching, arg11, arg12, ...)),
        asyncio.async(handle(s.searching, arg21, arg22, ...)),
        ...
        ]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

通过运行通过 pytest 测试这个测试用例，它通过了，但是搜索引擎会引发一些奇怪的异常。它说 Future/Task exception was never retrieved.

想问的事情：

对于我的第一次尝试，通过 return 函数调用的实际结果来使用 yield from 是正确的方法吗？
我想我需要为我的第二个测试用例添加一些睡眠以等待任务完成，但我应该怎么做？我怎样才能在我的第二个测试用例中获得对 return 的函数调用？
通过创建异步处理程序来处理请求，这是使用现有模块实现异步的好方法吗？
如果问题 2 的答案是否定的，是否每个客户端调用 class search 都需要包含 loop = get_event_loop() 这种东西来异步请求？

Answer 1

问题是您不能像 asyncio.coroutine 一样调用现有的同步代码并获得异步行为。当您调用 yield from searching(...) 时，如果 searching 本身实际上是一个 asyncio.coroutine，或者至少 return 是一个 asyncio.Future，您只会获得异步行为。现在，searching 只是一个常规的同步函数，所以调用 yield from searching(...) 只会抛出一个错误，因为它没有 return 一个 Future 或协程。

要获得您想要的行为，除了 synchronous 版本之外，您还需要 searching 的异步版本（或者如果您不这样做，则完全放弃同步版本需要它）。您有几个选项可以同时支持两者：

将 searching 重写为 asyncio.coroutine，它使用 asyncio 兼容调用来执行其 I/O，而不是阻塞 I/O。这将使它在 asyncio 上下文中工作，但这意味着您将无法再在同步上下文中直接调用它。相反，您还需要提供另一种同步 searching 方法来启动 asyncio 事件循环并调用 return loop.run_until_complete(self.searching(...))。有关详细信息，请参阅。

保持 searching 的同步实现，并提供使用 BaseEventLoop.run_in_executor 到运行的替代异步 API searching 后台线程中的方法：

class search(object):
  ...
  self.s = some_search_engine()
  ...
  def searching(self, *args, **kwargs):
    ret = {}
    ...
    return ret

   @asyncio.coroutine
   def searching_async(self, *args, **kwargs):
      loop = kwargs.get('loop', asyncio.get_event_loop())
      try:
          del kwargs['loop']  # assuming searching doesn't take loop as an arg
      except KeyError:
          pass
      r = yield from loop.run_in_executor(None, self.searching, *args)  # Passing None tells asyncio to use the default ThreadPoolExecutor
      return r

测试脚本：

s = search()
loop = asyncio.get_event_loop()
loop.run_until_complete(s.searching_async(arg1, arg2, ...))
loop.close()

这样，您可以保持同步代码不变，并且至少提供可以在 asyncio 代码中使用的方法，而不会阻塞事件循环。如果您在代码中实际使用异步 I/O，这不是一个干净的解决方案，但总比没有好。

提供两个完全独立的 searching 版本，一个使用阻塞 I/O，另一个 asyncio 兼容。这为两种上下文提供了理想的实现，但需要两倍的工作。

Python asyncio 任务的产出很差

Python asyncio task got bad yield

python

asynchronous

python-asyncio