Python，调用进程池而不阻塞事件循环 [英] Python, invoke a process pool without blocking the event loop

问题描述

如果我运行以下代码：

import asyncio
import time
import concurrent.futures

def cpu_bound(mul):
    for i in range(mul*10**8):
        i+=1
    print('result = ', i)
    return i

async def say_after(delay, what):
    print('sleeping async...')
    await asyncio.sleep(delay)
    print(what)

# The run_in_pool function must not block the event loop
async def run_in_pool():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        result = executor.map(cpu_bound, [1, 1, 1])

async def main():
    task1 = asyncio.create_task(say_after(0.1, 'hello'))
    task2 = asyncio.create_task(run_in_pool())
    task3 = asyncio.create_task(say_after(0.1, 'world'))

    print(f"started at {time.strftime('%X')}")
    await task1
    await task2
    await task3
    print(f"finished at {time.strftime('%X')}")

if __name__ == '__main__':
    asyncio.run(main())

输出为：

started at 18:19:28
sleeping async...
result =  100000000
result =  100000000
result =  100000000
sleeping async...
hello
world
finished at 18:19:34

这表明事件循环一直阻塞到CPU绑定作业（ task2 ）完成，然后继续执行 task3 。

This shows that the event loop blocks until the cpu bound jobs (task2) finish and it continues afterwards with the task3.

如果我只运行一个CPU作业（ run_in_pool 是以下项）：

If I run only one cpu bound job (the run_in_pool is the following one):

async def run_in_pool():
    loop = asyncio.get_running_loop()
    with concurrent.futures.ProcessPoolExecutor() as executor:
        result = await loop.run_in_executor(executor, cpu_bound, 1)

然后，由于输出为：

started at 18:16:23
sleeping async...
sleeping async...
hello
world
result =  100000000
finished at 18:16:28

如何运行许多cpu绑定的作业（在 task2 中）

How can I run many cpu bound jobs (in task2) in a process pool without blocking the event loop?

推荐答案

正如您发现的那样，您需要使用asyncio自己的 run_in_executor 等待提交的任务完成而不会阻塞事件循环。 Asyncio不提供与 map 等效的功能，但要模仿它并不难：

As you discovered, you need to use asyncio's own run_in_executor to wait for submitted tasks to finish without blocking the event loop. Asyncio doesn't provide the equivalent of map, but it's not hard to emulate it:

async def run_in_pool():
    with concurrent.futures.ProcessPoolExecutor() as executor:
        futures = [loop.run_in_executor(executor, cpu_bound, i)
                   for i in (1, 1, 1)]
        result = await asyncio.gather(*futures)

这篇关于Python，调用进程池而不阻塞事件循环的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！