使用multiprocessing+asyncio处理队列时queue.join()随机挂起问题

核心问题1：queue.qsize()的不可靠性+未捕获Empty异常 #

在多进程环境下，queue.qsize()返回的队列大小是瞬时值，当你调用qsize()后，其他进程可能已经取走了队列中的元素，此时用这个值循环获取元素，会出现实际队列元素不足的情况，导致queue.get(timeout=0.01)抛出queue.Empty异常。

代码中未捕获该异常，直接导致worker的async任务终止，那些已经被get()取出但还没调用task_done()的元素，会让队列的未完成任务计数永远无法归零，最终queue.join()一直阻塞。

核心问题2：同步get()阻塞async事件循环 #

multiprocessing.JoinableQueue的get()是同步阻塞方法，在async函数中调用会阻塞整个事件循环，导致同一进程内的其他worker无法执行，降低并发效率。

修复步骤 #

1. 安全批量获取元素，捕获异常 #

把批量获取逻辑改成循环尝试获取，直到拿到指定数量元素或队列为空，同时捕获Empty异常：

async def worker(i_process: int, queue: JoinableQueue):
    query_size = 200
    while True:
        batch = []
        while len(batch) < query_size:
            try:
                item = queue.get_nowait()  # 非阻塞获取，避免超时等待
                batch.append(item)
            except queue.Empty:
                break
        if not batch:
            break

        logging.info(f"p:{i_process} - processing batch of {len(batch)} items")
        await asyncio.sleep(0.01)  # 用异步操作模拟任务，避免阻塞事件循环

        for _ in batch:
            queue.task_done()

2. 适配asyncio，避免同步阻塞 #

如果实际处理任务是同步逻辑，应放到线程池执行，避免阻塞事件循环：

async def process_batch(batch):
    # 同步任务放入线程池执行
    loop = asyncio.get_running_loop()
    await loop.run_in_executor(None, lambda: [logging.info(f"Processing {item}") for item in batch])

async def worker(i_process: int, queue: JoinableQueue):
    query_size = 200
    while True:
        batch = []
        while len(batch) < query_size:
            try:
                item = queue.get_nowait()
                batch.append(item)
            except queue.Empty:
                break
        if not batch:
            break

        await process_batch(batch)

        for _ in batch:
            queue.task_done()

3. 确保进程正常退出 #

去掉return_exceptions=True，让异常直接抛出便于调试；任务完成后主动终止子进程：

async def async_process(i_process: int, queue: JoinableQueue):
    logging.info(f"p:{i_process} - launching workers...")
    workers = [asyncio.create_task(worker(i_process, queue)) for _ in range(N_WORKERS)]
    await asyncio.gather(*workers)

def start_processes(queue: JoinableQueue):
    processes = []
    for i in range(N_PROCESSES):
        p = Process(target=async_process_wrapper, args=(i, queue))
        p.start()
        processes.append(p)

    queue.join()
    # 终止并回收子进程
    for p in processes:
        p.terminate()
        p.join()

完整修复代码 #

import asyncio
import logging
import random
from multiprocessing import JoinableQueue, Process, Empty

N_PROCESSES = 4
N_WORKERS = 8

# 配置日志
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

def create_queue(zpids: list[int]) -> JoinableQueue:
    queue = JoinableQueue()
    for zpid in zpids:
        queue.put(zpid)
    return queue

async def process_batch(batch):
    # 模拟异步处理任务（同步任务可放入线程池）
    await asyncio.sleep(0.001)
    logging.info(f"Processed batch of {len(batch)} items")

async def worker(i_process: int, queue: JoinableQueue):
    query_size = 200
    while True:
        batch = []
        while len(batch) < query_size:
            try:
                item = queue.get_nowait()
                batch.append(item)
            except Empty:
                break
        if not batch:
            break

        await process_batch(batch)

        for _ in batch:
            queue.task_done()

async def async_process(i_process: int, queue: JoinableQueue):
    logging.info(f"p:{i_process} - launching workers...")
    workers = [asyncio.create_task(worker(i_process, queue)) for _ in range(N_WORKERS)]
    await asyncio.gather(*workers)

def async_process_wrapper(i_process: int, zpids: JoinableQueue):
    asyncio.run(async_process(i_process, zpids))

def start_processes(queue: JoinableQueue):
    processes = []
    for i in range(N_PROCESSES):
        p = Process(target=async_process_wrapper, args=(i, queue))
        p.start()
        processes.append(p)

    queue.join()
    for p in processes:
        p.terminate()
        p.join()

def main():
    data = [random.randrange(1, 1000) for _ in range(200000)]
    my_queue = create_queue(data)
    start_processes(my_queue)
    logging.info("All tasks completed!")

if __name__ == "__main__":
    main()

关键改进总结 #

• 用get_nowait()+捕获Empty异常替代不可靠的qsize()逻辑，避免worker因异常崩溃
• 同步任务放入线程池，避免阻塞async事件循环
• 去掉异常屏蔽，便于调试；任务完成后回收子进程，避免资源泄漏

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