如何在 asyncio 中使用阻塞函数 [英] How to use blocking functions with asyncio

问题描述

我在一个项目中使用 django ORM(在 django 之外).我的工作流程是

I'm using django ORM in a project(outside of django). My workflow is

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

问题是你不能在 async 环境中调用阻塞函数，你不能在阻塞环境中使用 async/await.

The problem is you can't call blocking functions in async environment and you can not use async/await in blocking environment.

我想出了两个解决方案:

I have come up with 2 solutions:

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

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

The whole program should be sync. And use asyncio.run()(Python 3.7) to call async functions needed.

我无法决定哪种方法更好.

I can't decide which one is better approach.

我知道之前有人问过类似的问题.我的问题是在尝试组合阻塞和非阻塞代码时是否有一般规则?

I know a similar question has been asked before. My question is is there a general rule when trying to combine blocking and non-blocking code?

推荐答案

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

Given the choice between those two, I would definitely recommend approach #1.

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

#2 has the downside that you're missing out on a lot of asyncio functionality by splitting up asyncio calls into separate little event loop runs. For example, you can't create a "background" task whose execution spans several calls to asyncio.run(), and that kind of thing can be very useful for logging, monitoring, or timeout. (Using asyncio.run could also be a performance issue because it creates a whole new event loop on every invocation, but this can be fixed by switching to run_until_complete.)

但还有第三种选择:

• 创建一个单独的线程，只执行loop.run_forever() 并等待被赋予的工作要做.程序的其余部分由普通的阻塞代码组成，可以使用 asyncio.run_coroutine_threadsafe().该函数不会阻塞；它立即返回一个 concurrent.futures.Future 你可以传递它，它的 result() 方法会自动等待结果可用.它支持其他功能，例如使用 wait，as_completed 迭代器等

• Create a separate thread that only executes loop.run_forever() and waits to be given work to do. The remainder of the program consists of normal blocking code that can request something from asyncio using asyncio.run_coroutine_threadsafe(). That function doesn't block; it immediately returns a concurrent.futures.Future which you can pass around and whose result() method automatically waits for the result to be available. It supports additional features, such as waiting for multiple instances to complete in parallel using wait, the as_completed iterator, etc.

恕我直言，这种方法结合了问题中两个选项的最佳特征.它让阻塞代码真正阻塞，仍然被允许等待事情发生，产生线程等，而不强制使用 async def 和 run_in_executor 一刀切.同时，可以使用 asyncio 最佳实践编写 asyncio 部分，并为整个程序提供长期运行的事件循环.你只需要注意all与应用程序其余部分的事件循环接口(甚至调用像loop.stop) 使用 loop.call_soon_threadsafe 和 asyncio.run_coroutine_threadsafe.

This approach IMHO combines the best characteristics of the two options from the question. It leaves blocking code truly blocking, still being allowed to wait for things to happen, spawn threads etc., without forcing the use of async def and run_in_executor across the board. At the same time the asyncio parts can be written using asyncio best practices, with a long-running event loop servicing the whole program. You just need to be careful for all interfacing with the event loop from the rest of the application (even to call something as simple as loop.stop) to be done using loop.call_soon_threadsafe and asyncio.run_coroutine_threadsafe.

这篇关于如何在 asyncio 中使用阻塞函数的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！