如何判断已读取multiprocessing Queue中的所有数据？

当tasks是没法提前数清楚个数的迭代器时，完全可以丢掉原来的结束标记和任务计数那套繁琐逻辑，换个更简洁的实现方式：

核心思路 #

1. 子进程只专心往队列写结果，不用再发什么结束标记
2. 主线程单独开个线程来读队列处理结果，避免卡着进程池的等待
3. 靠进程池的close()+join()等所有子进程跑完，这时候肯定不会有新结果进队列了
4. 最后让消费线程把队列里剩下的内容读完就完事

优化后的代码 #

from multiprocessing import Pool, Manager
import threading

def handle_task(arg, queue):
    # 子进程只负责写任务结果，不用管结束标记
    ... 这里写你的任务逻辑，往queue里塞结果 ...

def consume_queue(queue, done_flag):
    # 专门读队列的线程
    while not done_flag.is_set() or not queue.empty():
        try:
            value = queue.get(timeout=0.5)  # 加个超时，避免死等
            ... 这里写你处理结果的逻辑 ...
        except queue.Empty:
            continue

def main(tasks):
    manager = Manager()
    queue = manager.Queue()
    done_flag = threading.Event()

    # 先启动消费线程
    consumer = threading.Thread(target=consume_queue, args=(queue, done_flag))
    consumer.start()

    with Pool() as pool:
        # 用imap_unordered处理迭代器，不用提前知道任务数量
        pool.imap_unordered(handle_task, ((task, queue) for task in tasks))
        # 关闭进程池，不让加新任务
        pool.close()
        # 等所有子进程都干完活
        pool.join()

    # 告诉消费线程：所有任务都完了，读完剩下的就可以撤了
    done_flag.set()
    consumer.join()

为啥这方法更好？ #

• 不用提前搞任务数量，管你是迭代器还是啥都能处理
• 删掉了原来的标记符和计数逻辑，代码清爽多了
• 消费线程异步干活，主线程不用卡在队列读取上，效率更高

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