Asyncio CPU 绑定函数

Question

我有一个龙卷风网络应用程序，它大部分是异步的，但在迭代数据时有大量 CPU 绑定工作。大多数时候只有几百行，但有时可能有数万行，这会阻塞事件循环几秒钟。我想知道将这些同步函数转换为异步函数的首选方法。我的两个选择看起来像：

# Iterate over rows from db
i = 0
for row in rows:
    i += 1
    if i % 10000 == 0:
        await asyncio.sleep(0)
    # Do some processing

或

# func being the above code that iterates over rows without sleep
executor = concurrent.futures.ThreadPoolExecutor(max_workers=10)
await asyncio.get_event_loop().run_in_executor(executor, func)

我知道为了避免 GIL 限制，我应该使用 ProcessPoolExecutor，但大多数时候它必须迭代的行数非常小，因此使用 ProcessPoolExecutor 可能不值得花费这些开销。

Answer 1

这是一个可以在另一个线程中很好地运行的模式 - 即使使用 GIL，至少异步循环在处理过程中不会被阻塞。

此外，尝试使用 ProcessPoolExecutor 而不是默认的 ThreadPool 需要更改两行代码（其中一行是执行程序的

import

语句）。

asyncio.Loop.run_in_executor

返回一个可以等待的任务，独立的（仅阻止当前的Web视图功能），或者在专门的异步任务中运行，除了视图之外，如果方便的话。

现在，Python 3.13 的自由线程变体也值得一看 - ThreadpoolExecutor 可以免费并行运行，而不会产生使用 ProccessPoolExecutor 所固有的序列化开销。

总而言之，在 CPU 限制循环中不时运行

await asyncio.sleep(0)

的策略还不错，而且我时不时地使用过 - 但是，使用 threadpoolexecutor 没有任何缺点，你可以首先尝试一下。