如何在库函数中使用异步事件循环

Question

我正在尝试使用 asyncio 创建一个执行一些异步操作的函数，此函数的用户不需要知道 asyncio 是在幕后参与的。 我很难理解如何使用 asyncio API 完成此操作，因为大多数函数似乎都在使用 get_event_loop 访问的一些全局循环变量下运行，并且对此的调用受到此循环内的全局状态的影响。

我在这里有四个示例，其中两个（foo1 和 foo3）似乎是合理的用例，但它们都表现出非常奇怪的行为：

async def bar(loop):
    # Disregard how simple this is, it's just for example
    s = await asyncio.create_subprocess_exec("ls", loop=loop)


def foo1():
    # Example1: Just use get_event_loop
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000))
    # On exit this is written to stderr:
    #    Exception ignored in: <bound method BaseEventLoop.__del__ of <_UnixSelectorEventLoop running=False closed=True debug=False>>
    #    Traceback (most recent call last):
    #      File "/usr/lib/python3.5/asyncio/base_events.py", line 510, in __del__
    #      File "/usr/lib/python3.5/asyncio/unix_events.py", line 65, in close
    #      File "/usr/lib/python3.5/asyncio/unix_events.py", line 146, in remove_signal_handler
    #      File "/usr/lib/python3.5/signal.py", line 47, in signal
    #    TypeError: signal handler must be signal.SIG_IGN, signal.SIG_DFL, or a callable object


def foo2():
    # Example2: Use get_event_loop and close it when done
    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000))  # RuntimeError: Event loop is closed  --- if foo2() is called twice
    loop.close()


def foo3():
    # Example3: Always use new_event_loop
    loop = asyncio.new_event_loop()
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000)) #RuntimeError: Cannot add child handler, the child watcher does not have a loop attached
    loop.close()


def foo4():
    # Example4: Same as foo3 but also set_event_loop to the newly created one
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)        # Polutes global event loop, callers of foo4 does not expect this.
    loop.run_until_complete(asyncio.wait_for(bar(loop), 1000))  # OK
    loop.close()

这些功能都不起作用，我没有看到任何其他明显的方法来做到这一点，应该如何使用 asyncio？似乎它只是在假设应用程序的入口点是您可以创建和关闭全局循环的唯一地方的情况下使用的。我必须摆弄事件循环策略吗？

foo3 似乎是正确的解决方案，但即使我明确地传递循环，我也会得到一个错误，因为在 create_subprocess_exec 的深处，它正在使用当前策略来获得一个新的循环，即无，这是 asyncio 子进程中的错误吗？

我在 Ubuntu 上使用 Python 3.5.3。

Answer 1

foo1 错误是因为您没有关闭事件循环，请参阅此issue。

foo2 因为你不能重复使用封闭的事件循环。

foo3，因为您没有将新事件循环设置为全局。

foo4 几乎是你想要的，你剩下要做的就是存储旧的事件循环并在 bar 执行后将其设置回全局：

import asyncio


async def bar():
    # After you set new event loop global,
    # there's no need to pass loop as param to bar or anywhere else.
    process = await asyncio.create_subprocess_exec("ls")
    await process.communicate()


def sync_exec(coro):  # foo5
    old_loop = asyncio.get_event_loop()
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        loop.run_until_complete(coro)
    finally:
        loop.close()
        asyncio.set_event_loop(old_loop)


sync_exec(asyncio.wait_for(bar(), 1000))

更重要的一点：不清楚为什么要在某些同步函数后面隐藏使用 asyncio，但通常这是个坏主意。一个全局事件循环的全部内容是允许用户在这个单一事件循环中运行不同的并发作业。你正试图消除这种可能性。我认为你应该重新考虑这个决定。

Answer 2

升级到 Python 3.6，然后 foo1() 将工作，无需显式关闭默认事件循环。

不是我希望的答案，因为我们只使用 3.5 :(