将 asyncio 与多线程 ProcessPoolExecutor 和 async 结合使用

Question

我的问题非常类似于Combining asyncio with a multi-worker ProcessPoolExecutor - 但是一个细微的变化（我相信是async for）使得那里的优秀答案对我来说不可用。

我正在尝试以下 MWE：

import concurrent.futures
import asyncio
import time

async def mygen(u: int = 2):
    i = 0
    while i < u:
        yield i
        i += 1

def blocking(delay):
    time.sleep(delay+1)
    return('EXECUTOR: Completed blocking task number ' + str(delay+1))

async def non_blocking(loop):
    with concurrent.futures.ProcessPoolExecutor() as executor:
        async for i in mygen():
            print('MASTER: Sending to executor blocking task number ' + str(i+1))
            result = await loop.run_in_executor(executor, blocking, i)
            print(result)
            print('MASTER: Well done executor - you seem to have completed blocking task number ' + str(i+1))

loop = asyncio.get_event_loop()
loop.run_until_complete(non_blocking(loop))

正如预期的那样，此输出不是异步的：

MASTER: Sending to executor blocking task number 1
EXECUTOR: Completed blocking task number 1
MASTER: Well done executor - you seem to have completed blocking task number 1
MASTER: Sending to executor blocking task number 2 
EXECUTOR: Completed blocking task number 2 
MASTER: Well done executor - you seem to have completed blocking task number 2

我想调整代码，以便任务在两个并发进程中运行，并在可用时打印输出。期望的输出是：

MASTER: Sending to executor blocking task number 1
MASTER: Sending to executor blocking task number 2
EXECUTOR: Completed blocking task number 1
MASTER: Well done executor - you seem to have completed blocking task number 1
EXECUTOR: Completed blocking task number 2
MASTER: Well done executor - you seem to have completed blocking task number 2

我从将 asyncio 与多线程 ProcessPoolExecutor 相结合中了解到，就目前情况而言，我的语法await loop.run_in_executor()是阻塞的。我不知道如何以一种允许async for移动到下一个生成值的方式替换它，同时等待执行者完成他们的工作。注意我没有asyncio.gather像他们的例子那样使用。

Answer 1

如果您希望最多有两个进程运行您的任务，实现此目的的最简单方法是使用max_workers=2. 然后您可以尽快提交任务，即继续下一个迭代，而async for无需等待上一个任务完成。您可以在最后收集所有任务的结果，以确保异常不会被忽视（并可能获得实际结果）。

以下代码产生预期的输出：

from concurrent.futures import ProcessPoolExecutor
import asyncio
import time

async def mygen(u: int = 2):
    i = 0
    while i < u:
        yield i
        i += 1

def blocking(delay):
    time.sleep(delay+1)
    return('EXECUTOR: Completed blocking task number ' + str(delay+1))

async def run_blocking(executor, task_no, delay):
    print('MASTER: Sending to executor blocking task number '
          + str(task_no))
    result = await loop.run_in_executor(executor, blocking, delay)
    print(result)
    print('MASTER: Well done executor - you seem to have completed '
          'blocking task number ' + str(task_no))

async def non_blocking(loop):
    tasks = []
    with ProcessPoolExecutor(max_workers=2) as executor:
        async for i in mygen():
            # spawn the task and let it run in the background
            tasks.append(asyncio.create_task(
                run_blocking(executor, i + 1, i)))
        # if there was an exception, retrieve it now
        await asyncio.gather(*tasks)

loop = asyncio.get_event_loop()
loop.run_until_complete(non_blocking(loop))