如何在协议循环中将协程打包为普通函数？

Question

我正在使用asyncio作为网络框架.

在下面的代码中(low_level是我们的低级函数,mainblock是我们的程序入口,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(compatibility等等),但是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)return2)使用yield from.该asyncio.sleep呼叫low_level将永远无法完成,除非你做一个这两件事.

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

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"

此外,在您的示例代码,您应该使用@asyncio.coroutine装饰两个low_level和user_func,如在规定asyncio的文档:

协程是遵循某些约定的生成器.出于文档目的,所有协同程序都应该使用@ asyncio.coroutine进行修饰,但这不能严格执行.

编辑:

以下是来自同步Web框架的用户如何在不阻止其他请求的情况下调用您的应用程序:

@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,它将阻塞直到请求完成,但是应该为正在进行的所有异步I/O释放GIL low_level,允许在单独的线程中处理其他请求.