Python多进程Queue.empty()函数失效，无法完整取出队列元素

你遇到的核心问题是multiprocessing.Queue的empty()和qsize()方法在多进程环境下不具备原子性，返回结果不可靠——哪怕所有子进程已经终止，队列的内部缓冲可能还没完成同步，导致empty()误判为True，此时队列中仍有未读取的元素，最终导致res_list没有收集到全部结果。

哪怕循环因empty()返回True而退出，也不代表队列真的为空，这就是为什么每次运行结果都不一致，偶尔才符合预期。

方案1：已知元素总数，直接循环固定次数 #

如果你明确知道要收集的元素总数（比如你这里的16个），最直接的方式是循环对应次数调用get()，完全不需要依赖empty()判断：

import multiprocessing

result = multiprocessing.Queue(maxsize=16)

# 向队列中填充16个元素的多进程部分
# 注意：这里必须先等待所有子进程终止，比如用process.join()

res_list = []
# 循环16次，确保取出所有元素
for _ in range(16):
    res_list.append(result.get())

# 此时res_list长度必然是16
print(len(res_list))

方案2：使用JoinableQueue（推荐） #

multiprocessing.JoinableQueue是普通Queue的扩展，它支持task_done()和join()方法，能更可靠地跟踪所有任务的完成状态：

import multiprocessing

# 用JoinableQueue替代普通Queue
result = multiprocessing.JoinableQueue(maxsize=16)

def worker(item):
    # 处理你的业务逻辑
    result.put(item)
    # 标记当前任务已完成
    result.task_done()

# 创建并启动子进程
processes = []
for i in range(16):
    p = multiprocessing.Process(target=worker, args=(i,))
    p.start()
    processes.append(p)

# 等待所有子进程执行完毕
for p in processes:
    p.join()

# 等待队列中所有任务都被标记完成（确保所有元素都已放入队列）
result.join()

# 取出所有元素
res_list = []
while True:
    try:
        # 非阻塞取元素，超时时间设短即可
        item = result.get(block=False)
        res_list.append(item)
    except multiprocessing.Queue.Empty:
        break

print(len(res_list))  # 输出16

方案3：添加结束标记 #

如果无法提前知道元素总数，可以在所有子进程终止后，向队列中放入一个特殊的结束标记（比如None），主进程取到该标记时停止读取：

import multiprocessing

# 队列容量多留一个位置放结束标记
result = multiprocessing.Queue(maxsize=17)

def worker(item):
    # 处理业务逻辑
    result.put(item)

# 创建并启动子进程
processes = []
for i in range(16):
    p = multiprocessing.Process(target=worker, args=(i,))
    p.start()
    processes.append(p)

# 等待所有子进程终止
for p in processes:
    p.join()

# 放入结束标记
result.put(None)

# 读取元素直到遇到结束标记
res_list = []
while True:
    item = result.get()
    if item is None:
        break
    res_list.append(item)

print(len(res_list))  # 输出16

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