Python中如何正确结合ProcessPoolExecutor使用Queue？程序阻塞求助

问题描述 #

尝试通过Queue加载大量任务，用ProcessPoolExecutor创建进程池让每个进程从队列取任务处理，但程序陷入阻塞：队列已正确填充，但进程未执行worker子例程，输出以下内容后停滞：

running main
Size of queue 30

核心原因 #

1. 队列类型错误：使用了queue.Queue，这是线程安全队列，仅适用于线程间通信，不支持跨进程共享。ProcessPoolExecutor创建的子进程拥有独立内存空间，主进程的队列无法被子进程访问，子进程中的队列为空，导致worker无法进入任务处理逻辑。
2. 任务结束逻辑失效：原代码依赖queue.Queue的unfinished_tasks判断循环，但跨进程场景下该机制完全失效，同时主进程的q.join()会一直等待任务完成，形成死锁。

修正代码 #

使用multiprocessing.JoinableQueue（专为跨进程通信设计），并通过结束标记控制worker退出：

# -*- coding: utf-8 -*-
"""
Created on Tue Aug 30 17:08:42 2022

@author: Rahul
"""

from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
import time


def worker(id, q):
    print(f'{id}:: Worker running', flush=True)
    while True:
        item = q.get()
        # 收到结束标记则退出循环
        if item is None:
            q.task_done()
            break
        print(f'{id}::Working on {item}', flush=True)
        print(f'{id}::Finished {item}', flush=True)
        q.task_done()
        print(f'{id}::Sleeping. Item: {item}', flush=True)
        time.sleep(0.1)
    print(f'{id}:: Worker exited', flush=True)


def main():
    print('running main')
    # 使用跨进程的JoinableQueue
    q = mp.JoinableQueue()
    
    # 填充30个任务
    for item in range(30):
        q.put(item)
    
    # 给每个worker添加结束标记（4个worker对应4个None）
    for _ in range(4):
        q.put(None)
    
    print(f'Size of queue {q.qsize()}')

    # 启动进程池并分配任务
    with ProcessPoolExecutor(max_workers=4) as executor:
        executor.map(worker, [1,2,3,4], [q]*4)

    # 等待所有任务处理完成
    q.join()
    print('All work completed')


if __name__ == "__main__":
    main()

关键修正点 #

• 替换queue.Queue为mp.JoinableQueue：实现跨进程的队列共享，支持task_done()和join()方法跟踪任务完成状态。
• 添加结束标记：通过向队列放入与worker数量一致的None，让每个worker在处理完所有任务后主动退出循环，避免无限等待。
• 确保队列共享：进程池中的worker拿到的是同一个跨进程队列实例，而非主进程队列的副本。

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