asyncio run or run_until_complete

Question

asyncio run or run_until_complete

39

我正在以一种非常基础的方式使用asyncio应用程序。查看互联网上的大多数教程（甚至官方文档），我发现它们使用get_event_loop()和loop.run_until_complete()：

import asyncio

async def say(what, when):
    await asyncio.sleep(when)
    print(what)

loop = asyncio.get_event_loop()
loop.run_until_complete(say('hello world', 1))
loop.close()

但是在Python 3.7文档中，我们可以读到：

应用程序开发者通常应使用高级asyncio函数，例如asyncio.run(), 很少需要引用循环对象或调用其方法。此部分主要面向较低级别代码、库和框架的作者，他们需要更精细地控制事件循环行为。

我发现这种方法更加简洁易用，但它仅适用于Python 3.7以上版本。因此，我必须做出选择，是使用Python 3.7+和run()还是使其与Python 3.6兼容并使用事件循环。你会如何处理这个问题？有没有简单的方法使它向后兼容Python 3.6？我应该先检查Python版本，然后根据版本使用其中一种方式，直到Python 3.7成为常见版本？

- makeMonday

2

如果你要编写既适用于较旧版本又适用于新版本的更复杂的代码，并且你会在它们之间动态切换...那么选择坚持使用既适用于较旧版本又适用于新版本的更复杂的代码不是更容易吗？ - deceze

1

@deceze 是的，也许那是最好的选择，我想听听您的意见，并在使其兼容的情况下知道最佳方法是什么。 - makeMonday

1

@deceze 在旧版本的Python上模拟asyncio.run并不难，而且你可以获得一个优势，即在由asyncio.run设置的条件下测试你的代码，也就是在一个新创建的事件循环中。 - user4815162342

2个回答

4

你可以通过从asyncio.runners.py中复制代码来复制asyncio.run功能。下面的代码是Python 3.8的版本。

from asyncio import coroutines, events, tasks


def run(main, *, debug=False):
    """Execute the coroutine and return the result.

    This function runs the passed coroutine, taking care of
    managing the asyncio event loop and finalizing asynchronous
    generators.

    This function cannot be called when another asyncio event loop is
    running in the same thread.

    If debug is True, the event loop will be run in debug mode.

    This function always creates a new event loop and closes it at the end.
    It should be used as a main entry point for asyncio programs, and should
    ideally only be called once.

    Example:

        async def main():
            await asyncio.sleep(1)
            print('hello')

        asyncio.run(main())
    """
    if events._get_running_loop() is not None:
        raise RuntimeError(
            "asyncio.run() cannot be called from a running event loop")

    if not coroutines.iscoroutine(main):
        raise ValueError("a coroutine was expected, got {!r}".format(main))

    loop = events.new_event_loop()
    try:
        events.set_event_loop(loop)
        loop.set_debug(debug)
        return loop.run_until_complete(main)
    finally:
        try:
            _cancel_all_tasks(loop)
            loop.run_until_complete(loop.shutdown_asyncgens())
        finally:
            events.set_event_loop(None)
            loop.close()


def _cancel_all_tasks(loop):
    to_cancel = tasks.all_tasks(loop)
    if not to_cancel:
        return

    for task in to_cancel:
        task.cancel()

    loop.run_until_complete(
        tasks.gather(*to_cancel, loop=loop, return_exceptions=True))

    for task in to_cancel:
        if task.cancelled():
            continue
        if task.exception() is not None:
            loop.call_exception_handler({
                'message': 'unhandled exception during asyncio.run() shutdown',
                'exception': task.exception(),
                'task': task,
            })

- Ted Petrou

1

在版本3.9中，他们在shutdown_asyncgens之后添加了 loop.run_until_complete(loop.shutdown_default_executor())（除此之外相同），而在3.10中没有任何更改。 - Louis Maddox

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- user4815162342 · Accepted Answer

如何使使用asyncio.run的代码兼容Python 3.6？

您可以实现一个简单的替代asyncio.run的函数，并在较旧的Python版本上调用它：

import asyncio, sys, types

def run(coro):
    if sys.version_info >= (3, 7):
        return asyncio.run(coro)

    # Emulate asyncio.run() on older versions

    # asyncio.run() requires a coroutine, so require it here as well
    if not isinstance(coro, types.CoroutineType):
        raise TypeError("run() requires a coroutine object")

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()
        asyncio.set_event_loop(None)

使用这种方法的优势在于，即使在旧版本的Python上运行代码，也能够接近新asyncio.run的语义方式执行代码。(例如，您将始终在新创建的事件循环上运行)。放弃对Python 3.7以前版本的支持将像删除run shim并直接调用asyncio.run一样容易。