如何将协程打包为事件循环中的普通函数？

Question

我正在使用 asyncio 来搭建一个网络框架。

在下面的代码中（low_level 是我们低级别的函数，main 块是我们程序的入口，user_func 是用户自定义的函数）：

import asyncio

loop = asyncio.get_event_loop()
""":type :asyncio.AbstractEventLoop"""


def low_level():
    yield from asyncio.sleep(2)


def user_func():
    yield from low_level()


if __name__ == '__main__':
    co = user_func()
    loop.run_until_complete(co)

我想把 low_level 包装成一个普通的函数，而不是 coroutine（为了兼容性等原因），但是 low_level 是在事件循环中运行的。我该如何把它包装成一个普通的函数呢？

Answer 1

因为 low_level 是一个协程，它只能在运行 asyncio 事件循环的情况下使用。如果你想在没有运行事件循环的同步代码中调用它，就需要提供一个包装器，这个包装器会启动一个事件循环，并运行协程直到完成：

def sync_low_level():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(low_level())

如果你想从一个已经在运行事件循环中的函数调用 low_level()，让它阻塞两秒，但又不想使用 yield from，那么答案是你不能这么做。事件循环是单线程的；每当你的函数在执行时，事件循环就会被阻塞，其他的事件或回调都无法处理。让在事件循环中运行的函数将控制权交回事件循环的唯一方法是 1) return 2) 使用 yield from。在 low_level 中的 asyncio.sleep 调用，除非你做了这两件事，否则永远无法完成。

现在，我想你可以创建一个全新的事件循环，并用它来从作为默认事件循环一部分的协程中同步运行睡眠：

import asyncio

loop = asyncio.get_event_loop()

@asyncio.coroutine
def low_level(loop=None):
    yield from asyncio.sleep(2, loop=loop)


def sync_low_level():
    new_loop = asyncio.new_event_loop()
    new_loop.run_until_complete(low_level(loop=new_loop))

@asyncio.coroutine
def user_func():
    sync_low_level()

if __name__ == "__main__":
    loop.run_until_complete(user_func())

但我真的不明白你为什么想这么做。

如果你只是想让 low_level 像一个返回 Future 的方法那样工作，以便你可以给它附加回调等，只需将它包装在 asyncio.async() 中：

loop = asyncio.get_event_loop()

def sleep_done(fut):
    print("Done sleeping")
    loop.stop()

@asyncio.coroutine
def low_level(loop=None):
    yield from asyncio.sleep(2, loop=loop)

def user_func():
    fut = asyncio.async(low_level())
    fut.add_done_callback(sleep_done)

if __name__ == "__main__":
    loop.call_soon(user_func)
    loop.run_forever()

输出：

<2 second delay>
"Done sleeping"

另外，在你的示例代码中，你应该对 low_level 和 user_func 都使用 @asyncio.coroutine 装饰器，正如 asyncio 文档中所述：

协程是遵循某些约定的生成器。为了文档目的，所有协程都应该使用 @asyncio.coroutine 装饰器，但这不能严格执行。

编辑：

以下是一个来自同步网络框架的用户如何在不阻塞其他请求的情况下调用你的应用程序：

@asyncio.coroutine
def low_level(loop=None):
    yield from asyncio.sleep(2, loop=loop)

def thr_low_level():
   loop = asyncio.new_event_loop()
   t = threading.Thread(target=loop.run_until_complete, args(low_level(loop=loop),))
   t.start()
   t.join()

如果一个由 Flask 处理的请求调用 thr_low_level，它会阻塞直到请求完成，但全局解释器锁（GIL）应该会释放，以便在 low_level 中进行的所有异步 I/O 允许其他请求在不同的线程中处理。