Python queue.SimpleQueue 怎样在生产者消费者间传递任务

如果只是在线程之间传递任务，生产者用 SimpleQueue.put() 写入，消费者用 SimpleQueue.get() 读取即可。它是线程安全的无界 FIFO 队列，空队列上的 get() 会等待；代价是没有容量上限，也没有 task_done() 和 join() 这样的任务完成追踪。

因此，SimpleQueue 适合“先把任务安全交给另一个线程”的轻量场景。如果生产速度可能长期高于消费速度，或者主线程必须等待全部任务处理完，应改用带 maxsize 的 queue.Queue。

最小可靠写法是：生产者只负责 put，消费者阻塞 get；全部任务写完后再放入一个进程内唯一的哨兵对象，让消费者明确退出。不要用 empty() 判断“工作结束”，也不要把 SimpleQueue 当成有背压的任务队列。

先看清 SimpleQueue 解决什么问题

Python 的 queue 模块面向多生产者、多消费者的线程通信。SimpleQueue 只保留最核心的 FIFO 传递能力：任务进入队列后，消费者按照进入顺序取出。它从 Python 3.7 起提供，容量不受 maxsize 控制。

这里的“无界”不是任务永远不会占内存，而是队列接口不会因为达到容量上限而暂停生产者。生产速度持续超过消费速度时，积压对象仍会消耗内存，所以容量控制是它与 Queue 的关键区别。

最小配方：生产者 put，消费者 get

生产者不需要自己加锁，直接把任务对象放入队列；消费者调用默认阻塞的 get()，队列为空时等待新任务。结束时使用一个只在当前进程中可见的 object() 作为哨兵，避免真实任务值与结束标记混淆。

import queue
import threading

tasks = queue.SimpleQueue()
STOP = object()  # 用唯一对象表示“生产结束”，避免和真实任务值冲突


def producer():
    for task_id in range(3):
        tasks.put({"id": task_id, "payload": f"job-{task_id}"})  # 线程安全地写入 FIFO 队列
    tasks.put(STOP)  # 所有任务写完后再发送结束标记


def consumer():
    while True:
        task = tasks.get()  # 队列为空时等待，不用轮询 empty()
        if task is STOP:
            break  # 只让消费者退出，不把哨兵当成业务任务处理
        print(f"处理任务：{task['payload']}")  # 实际项目中在这里调用业务函数


worker = threading.Thread(target=consumer)
worker.start()
producer()  # 示例中由当前线程充当生产者
worker.join()  # 等待消费者识别哨兵并结束

这个例子中，get() 的等待由队列内部完成，消费者不需要写“先判断为空、再睡眠、再重试”的循环。STOP 必须在所有真实任务之后写入；如果有多个消费者，就要放入与消费者数量相同的哨兵，确保每个消费者都能收到退出信号。

关键接口的边界

SimpleQueue.put(item, block=True, timeout=None) 的参数形状与 Queue.put() 兼容，但它不会因为队列容量而等待，block 和 timeout 不会提供背压语义。不要仅仅把 timeout=1 写上，就以为生产者会在一秒后放弃。

get(block=True, timeout=None) 默认等待任务。需要在没有任务时立即返回，可以使用 get_nowait()，但这会引入 queue.Empty 的分支；连续轮询通常比等待更浪费，也容易让退出条件变复杂。

qsize() 和 empty() 只能作为近似观察值。在多线程竞争下，检查结果与下一次操作之间可能已经发生变化，所以它们不能证明“此刻没有任务”或“后续 get 一定不阻塞”。

没有 task_done 和 join 时怎样收尾

SimpleQueue 不维护“已放入但尚未完成”的任务计数，因此不能像 Queue 那样调用 task_done()，也不能用 join() 等待队列任务全部完成。轻量场景可以用线程的 join() 等待消费者线程结束，但前提是消费者有明确的哨兵退出路径。

如果业务失败也必须计数、重试或汇总，建议把结果和异常状态显式写入另一个结果队列，或者直接使用 Queue 的任务追踪能力。不要假设“消费者线程还活着”就等于所有任务都已处理成功。

SimpleQueue 和 Queue 怎么选

常见误区

• 把无界当成无限资源：SimpleQueue 不阻塞生产者，不代表积压没有内存成本。
• 用 empty() 作为结束条件：队列暂时为空可能只是生产者还没来得及放入下一项，应该用哨兵或独立停止事件表达生命周期。
• 给 put() 加 timeout 期待限流：SimpleQueue 没有容量槽位，限流应交给 Queue、信号量或业务层速率控制。
• 忘记多个消费者的哨兵数量：每个需要退出的消费者都应收到一个结束标记，或采用单独的停止协议。

小结

SimpleQueue 的核心价值是把线程安全的 FIFO 传递压缩成 put() 和 get() 两个动作。生产者消费者模型中，用阻塞 get() 等待工作，用唯一哨兵表达生产结束；如果需要限制积压或等待业务完成，就把选择切换到 Queue。

相关问题

• SimpleQueue 的 get() 为什么比轮询 empty() 更适合消费者线程？
• 多个消费者共享一个 SimpleQueue 时，结束哨兵应该放几个？
• 什么时候应该用 Queue(maxsize=n) 给生产者施加背压？