如何通过 asyncio 使用阻塞函数

Question

我在一个项目中使用 django ORM（在 django 之外）。我的工作流程是

1. 通过 django ORM 选择对象并
2. 然后使用 asyncio lib 将其发送到消息队列

问题是您无法在异步环境中调用阻塞函数，并且无法在阻塞环境中使用 async/await 。

我想出了两个解决方案：

1. 整个程序应该是异步的。并loop.run_in_executor在需要时用于调用阻塞函数。

2. 整个程序应该是同步的。并使用asyncio.run()（Python 3.7）调用所需的异步函数。

我无法决定哪一种方法更好。

我知道以前也有人问过类似的问题。我的问题是尝试组合阻塞和非阻塞代码时是否有一般规则？

Answer 1

考虑到这两者之间的选择，我肯定会推荐方法#1。

#2 的缺点是，通过将异步调用拆分为单独的小事件循环运行，您会错过许多异步功能。例如，您不能创建一个“后台”任务，其执行跨越对 的多次调用asyncio.run()，并且这种事情对于日志记录、监视或超时非常有用。（使用asyncio.run也可能是一个性能问题，因为它在每次调用时都会创建一个全新的事件循环，但这可以通过切换到 来解决run_until_complete。）

但还有第三种选择：

• 创建一个单独的线程，仅执行loop.run_forever()并等待分配工作。程序的其余部分由正常的阻塞代码组成，可以使用asyncio.run_coroutine_threadsafe(). 该函数不会阻塞；它立即返回一个concurrent.futures.Future可以传递的对象，并且它的result()方法会自动等待结果可用。wait它支持附加功能，例如使用、as_completed迭代器等等待多个实例并行完成。

恕我直言，这种方法结合了问题中两个选项的最佳特征。它使阻塞代码真正阻塞，仍然允许等待事情发生，产生线程等，而不强制全面使用async def和run_in_executor。同时，可以使用异步最佳实践来编写异步部分，并使用长期运行的事件循环为整个程序提供服务。您只需要小心使用和来完成与应用程序其余部分的事件循环的所有接口（甚至调用像 一样简单的东西） 。loop.stoploop.call_soon_threadsafeasyncio.run_coroutine_threadsafe