如何使用ProcessPoolExecutor和run_in_executor在asyncio中处理队列？

你的代码已经实现了异步生产者+进程池处理CPU密集任务的核心逻辑，整体思路是正确的，但存在几个可以优化和补充的细节点，以下是具体建议：

1. 完善队列任务生命周期管理 #

当前process_queue中提交任务到进程池后，没有等待任务完成，也未调用aqueue.task_done()，这会导致队列无法追踪任务的完成状态。如果后续需要等待所有队列任务处理完毕再执行收尾操作（比如资源清理），会出现逻辑漏洞。

修改方式：等待run_in_executor返回的Future完成，再标记任务完成：

async def process_queue(executor):
    loop = asyncio.get_event_loop()
    while True:
        item = await aqueue.get()
        if item is None:
            aqueue.task_done()  # 标记终止信号的任务完成
            break
        try:
            await loop.run_in_executor(executor, process_item, item)
        finally:
            aqueue.task_done()  # 无论成功失败，都标记任务完成

同时，在supervisor中，等待队列所有任务完成后再结束：

async def supervisor():
    with ProcessPoolExecutor() as executor:
        processor_task = asyncio.create_task(process_queue(executor))
        await asyncio.gather(*[produce_item(i) for i in range(100)])
        await aqueue.put(None)
        await aqueue.join()  # 等待队列中所有任务处理完成
        await processor_task

2. 显式配置进程池大小 #

ProcessPoolExecutor默认使用os.cpu_count()个进程，但CPU密集型任务的最优进程数通常等于CPU核心数（避免上下文切换开销）。如果需要限制资源或适配特定场景，建议显式指定max_workers：

with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
    # ...

3. 避免全局变量，提升代码可维护性 #

当前aqueue和sem是全局变量，在小型示例中没问题，但在大型项目中会降低代码的复用性和可测试性。建议将这些变量封装到supervisor中，通过参数传递给生产者和消费者：

async def produce_item(item, queue, sem):
    async with sem:
        print(f"produce {item}")
        await asyncio.sleep(random() * 2)
        await queue.put(item)

async def process_queue(executor, queue):
    loop = asyncio.get_event_loop()
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        try:
            await loop.run_in_executor(executor, process_item, item)
        finally:
            queue.task_done()

async def supervisor():
    aqueue = asyncio.Queue()
    sem = asyncio.Semaphore(10)
    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        processor_task = asyncio.create_task(process_queue(executor, aqueue))
        await asyncio.gather(*[produce_item(i, aqueue, sem) for i in range(100)])
        await aqueue.put(None)
        await aqueue.join()
        await processor_task

4. 异常处理与优雅退出 #

• 生产者任务如果抛出异常，会导致asyncio.gather失败，可能无法发送终止信号None，导致process_queue一直阻塞。建议在gather中添加return_exceptions=True，或者捕获异常确保终止信号能被发送：

try:
    await asyncio.gather(*[produce_item(i, aqueue, sem) for i in range(100)], return_exceptions=True)
finally:
    await aqueue.put(None)

• 进程池中的process_item如果抛出异常，当前代码会静默失败。可以在process_queue中捕获异常并处理（比如日志记录）：

try:
    await loop.run_in_executor(executor, process_item, item)
except Exception as e:
    print(f"Failed to process item {item}: {e}")
finally:
    aqueue.task_done()

5. 任务结果收集（可选） #

如果需要收集process_item的处理结果，可以将结果存入另一个队列或列表：

async def process_queue(executor, queue, result_queue):
    loop = asyncio.get_event_loop()
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            await result_queue.put(None)
            break
        try:
            result = await loop.run_in_executor(executor, process_item, item)
            await result_queue.put((item, result))
        except Exception as e:
            await result_queue.put((item, e))
        finally:
            queue.task_done()

优化后的完整代码 #

import asyncio
import os
import time
from concurrent.futures import ProcessPoolExecutor
from random import random

async def produce_item(item, queue, sem):
    async with sem:
        print(f"produce {item}")
        await asyncio.sleep(random() * 2)  # simulate waiting for i/o
        await queue.put(item)

async def process_queue(executor, queue):
    loop = asyncio.get_event_loop()
    while True:
        item = await queue.get()
        if item is None:
            queue.task_done()
            break
        try:
            await loop.run_in_executor(executor, process_item, item)
        except Exception as e:
            print(f"Error processing item {item}: {str(e)}")
        finally:
            queue.task_done()

def process_item(item):
    print(f"process {item}, pid {os.getpid()}")
    time.sleep(random())  # simulate long processing time
    return f"Result for {item}"

async def supervisor():
    aqueue = asyncio.Queue()
    sem = asyncio.Semaphore(10)
    with ProcessPoolExecutor(max_workers=os.cpu_count()) as executor:
        processor_task = asyncio.create_task(process_queue(executor, aqueue))
        try:
            await asyncio.gather(*[produce_item(i, aqueue, sem) for i in range(100)], return_exceptions=True)
        finally:
            await aqueue.put(None)
        await aqueue.join()
        await processor_task

if __name__ == "__main__":
    asyncio.run(supervisor())

内容的提问来源于stack exchange，提问作者Bruce Johnson