Python线程队列问题：无法接收最后一个任务结果

问题根源分析 #

1. 线程循环条件存在竞态问题：while not self.queue_in.empty()在多线程场景下不可靠——当多个线程同时判断队列非空后，其中一个线程取走最后一个任务，剩余线程会在queue_in.get()处无限阻塞，无法执行后续的queue_out.put()，导致最后一个任务的结果无法存入，且线程无法正常退出，触发join()无限循环。
2. queue_out未完成任务标记：queue.Queue.join()要求队列中每个put的任务都对应调用task_done()来标记完成，否则会一直等待任务计数清零，直接引发无限循环。

修正方案及代码 #

核心修改点 #

• 改用无限循环+终止信号的线程退出逻辑，避免线程阻塞在get()操作上
• 为每个Worker线程添加对应数量的终止信号（如None），确保所有线程能正常退出
• 在消费queue_out结果时，调用task_done()标记任务完成，让queue_out.join()能正常结束

修正后完整代码 #

import threading
import queue

def insert_rows(task):
    # 模拟数据库插入逻辑，需替换为实际业务代码
    return f"table_{task}", len(task)

class Worker(threading.Thread):
    def __init__(self, queue_in, queue_out, **kwargs):
        super().__init__(**kwargs)
        self.queue_in = queue_in
        self.queue_out = queue_out

    def run(self):
        while True:
            task = self.queue_in.get()
            try:
                if task is None:
                    # 收到终止信号，退出循环
                    break
                # 处理数据库任务
                table_name, rows_inserted = insert_rows(task)
                # 将结果存入输出队列
                self.queue_out.put((table_name, rows_inserted))
            finally:
                # 确保队列任务计数正确减少，即使处理出错
                self.queue_in.task_done()

def do_db_stuff():
    queue_in = queue.Queue()
    queue_out = queue.Queue()
    files = ["file1", "file2", "file3", "file4"]  # 替换为实际文件列表

    # 放入所有业务任务
    for file in files:
        queue_in.put(file)

    thread_count = 3
    # 为每个线程添加终止信号，数量需与线程数一致
    for _ in range(thread_count):
        queue_in.put(None)

    # 创建并启动线程
    threads = [Worker(queue_in, queue_out) for _ in range(thread_count)]
    for t in threads:
        t.start()

    # 等待输入队列所有任务（包括终止信号）处理完成
    queue_in.join()

    # 等待所有线程完全退出，确保结果全部写入queue_out
    for t in threads:
        t.join()

    # 读取输出队列结果并标记任务完成
    results = []
    while not queue_out.empty():
        res = queue_out.get()
        results.append(res)
        queue_out.task_done()

    # 执行统计逻辑
    print("统计结果:", results)
    # 可选：等待输出队列所有任务标记完成
    queue_out.join()

if __name__ == "__main__":
    do_db_stuff()

关键细节说明 #

• 使用try...finally包裹queue_in.task_done()，避免任务处理出错时导致queue_in.join()无限等待
• 终止信号数量必须与线程数匹配，确保每个线程都能收到退出指令
• 主线程先等待输入队列完成，再等待线程退出，最后读取输出结果，彻底避免遗漏最后一个任务的结果

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