python - 等待异步功能完成

Question

我的问题或多或少类似于 this ，这实际上是一个 X-Y 问题，导致返回 this .然而，这不是重复的，因为我的用例略有不同，链接的线程没有回答我的问题。

我正在将一组同步程序从 Java 移植到 Python。这些程序与异步库交互。在 Java 中，我可以阻塞并等待该库的异步函数返回一个值，然后使用该值执行操作。

这是一个代码示例来说明问题。

def do_work_sync_1(arg1, arg2, arg3):
    # won't even run because await has to be called from an async function
    value = await do_work_async(arg1, arg2, arg3)

def do_work_sync_2(arg1, arg2, arg3):
    # throws "Loop already running" error because the async library referenced in do_work_async is already using my event loop
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(do_work_async(arg1, arg2, arg3))

def do_work_sync_3(arg1, arg2, arg3):
    # throws "got Future attached to a different loop" because the do_work_async refers back to the asynchronous library, which is stubbornly attached to my main loop
    thread_pool = ThreadPoolExecutor()
    future = thread_pool.submit(asyncio.run, do_work_async(arg1, arg2, arg3)
    result = future.result()

def do_work_sync_4(arg1, arg2, arg3):
    # just hangs forever
    event_loop = asyncio.get_event_loop()
    future = asyncio.run_coroutine_threadsafe(do_work_async(arg1, arg2, arg3), event_loop)
    return_value = future.result()

async def do_work_async(arg1, arg2, arg3):
    value_1 = await async_lib.do_something(arg1)
    value_2 = await async_lib.do_something_else(arg2, arg3)

    return value_1 + value_2

Python 似乎非常努力地阻止我在任何地方阻止任何事情。 await只能从 async def 开始使用函数，它们依次必须是 await编。似乎没有内置的方式来保持 async def/ await像病毒一样通过我的代码传播。
Task s 和 Future s 没有任何内置阻塞或 wait_until_complete机制，除非我想循环 Task.done() ，这看起来很糟糕。

我试过 asyncio.get_event_loop().run_until_complete() ，但这会产生错误: This event loop is already running. Apparently I'm not supposed to do that for anything except main() .

The second linked question above建议使用单独的线程并将异步函数包装在其中。我用一些简单的函数对此进行了测试，它似乎可以作为一个通用概念。这里的问题是我的异步库保留了对主线程事件循环的引用，并在我尝试从新线程引用它时抛出错误: got Future <Future pending> attached to a different loop .

我考虑将所有对异步库的引用移动到一个单独的线程中，但我意识到我仍然无法在新线程中阻塞，我必须创建第三个线程来阻塞调用，这将使我回到 Future attached to a different loop错误。

我在这里几乎没有想法。有没有办法阻止并等待异步函数返回，或者我真的被迫将整个程序转换为 async/ await ? (如果是后者，解释会很好。我不明白。)

Answer 1

我花了一些时间，但最后我找到了真正的问题😇

Is there a way to block and wait for an async function to return, or am I really being forced to convert my entire program to async/await?

有一个高级函数 asyncio.run() .它做了三件事:

创建新的事件循环

在该事件循环中运行您的异步函数

等待任何未完成的任务并关闭循环

它的源代码在这里: https://github.com/python/cpython/blob/3221a63c69268a9362802371a616f49d522a5c4f/Lib/asyncio/runners.py#L8你看它使用 loop.run_until_complete(main)在引擎盖下。

如果您正在编写完全异步的代码，您应该调用 asyncio.run()在您的 main() 末尾的某处功能，我猜。但也不必如此。您可以随心所欲地运行它，次数不限。注意事项:

在给定线程中，一次只能运行一个事件循环

不要从 async def 运行它函数，因为很明显，你已经有一个事件循环在运行，所以你可以使用 await 调用该函数。相反

例子:

import asyncio

async def something_async():
    print('something_async start')
    await asyncio.sleep(1)
    print('something_async done')

for i in range(3):
    asyncio.run(something_async())

您可以让多个线程拥有自己的事件循环:

import asyncio
import threading

async def something_async():
    print('something_async start in thread:', threading.current_thread())
    await asyncio.sleep(1)
    print('something_async done in thread:', threading.current_thread())

def main():
    t1 = threading.Thread(target=asyncio.run, args=(something_async(), ))
    t2 = threading.Thread(target=asyncio.run, args=(something_async(), ))
    t1.start()
    t2.start()
    t1.join()
    t2.join()

if __name__ == '__main__':
    main()

如果您遇到此错误: Future attached to a different loop这可能意味着两点:

您正在使用与另一个事件循环相关的资源，而不是现在正在运行

您在开始事件循环之前创建了一些资源 - 在这种情况下它使用“默认事件循环” - 但是当您运行时 asyncio.run() ，你开始一个不同的循环。我以前遇到过这个:asyncio.Semaphore RuntimeError: Task got Future attached to a different loop

您需要使用 Python 版本至少 3.5.3 - explanation here .