Python3 queue.Queue基本用法：创建、put与get操作示例

在 Python 多线程编程中，不同线程之间直接共享普通列表（list）等集合类型容易导致竞态条件（Race Condition）。标准库内置的 queue.Queue 专门用来解决多线程之间的数据传递与任务调度问题。它底层基于锁与条件变量实现了线程安全的先进先出（FIFO，First In First Out）机制，核心操作围绕创建队列、使用 put() 写入数据以及使用 get() 取出数据展开。

图 1：queue.Queue 先进先出与多线程存取基本原理

一、创建队列对象与容量控制

使用队列前需要导入标准库中的 queue 模块。在实例化 queue.Queue 时，可以通过参数 maxsize 指定队列能容纳的数据项上限：

• maxsize <= 0（默认值 0）：表示无界队列，容量受系统可用内存限制，写入操作不会因队列满而阻塞。
• maxsize > 0：表示有界队列，当队列内元素达到指定上限时，新的写入请求会进入等待状态，起到流量控制和背压保护的作用。

import queue

# 1. 创建默认的无界队列
unbounded_q = queue.Queue()

# 2. 创建容量上限为 5 的有界队列
bounded_q = queue.Queue(maxsize=5)

print("队列容量上限:", bounded_q.maxsize)  # 输出: 5

二、使用 put 向队列存入数据

方法 put(item, block=True, timeout=None) 用于向队尾插入数据。当队列存在容量上限且已满时，block 与 timeout 参数决定具体行为：

• block=True, timeout=None（默认）：线程阻塞等待，直到队列有空位才存入并返回。
• block=True, timeout=n：线程最多阻塞等待 n 秒；若超时队列仍处于满载状态，抛出 queue.Full 异常。
• block=False：非阻塞写入，如果队列已满立即抛出 queue.Full 异常。等价于直接调用 put_nowait(item)。

图 2：put 与 get 操作的阻塞等待、超时与异常触发路径

以下代码演示了容量满载时的写入与超时捕获机制：

import queue

q = queue.Queue(maxsize=2)

# 正常存入两个元素
q.put("任务A")
q.put("任务B")
print("当前元素数量:", q.qsize())  # 输出: 2

# 非阻塞尝试写入第三个元素
try:
    q.put_nowait("任务C")
except queue.Full:
    print("写入失败: 队列已满 (put_nowait)")

# 阻塞等待 1 秒尝试写入
try:
    q.put("任务C", block=True, timeout=1.0)
except queue.Full:
    print("写入失败: 等待 1 秒后队列仍然已满")

三、使用 get 从队列取出数据

方法 get(block=True, timeout=None) 从队首移除并返回一个数据项。当队列为空时，其参数控制读取行为：

• block=True, timeout=None：线程阻塞挂起，直到有其他线程调用 put() 放入数据为止。
• block=True, timeout=n：最多等待 n 秒，仍无可用数据时抛出 queue.Empty 异常。
• block=False：立即尝试读取，若队列为空则抛出 queue.Empty。等价于 get_nowait()。

import queue

q = queue.Queue()
q.put("数据1")

# 取出已有数据
item = q.get()
print("取出项:", item)  # 输出: 取出项: 数据1

# 此时队列为空，带超时读取
try:
    val = q.get(block=True, timeout=1.5)
except queue.Empty:
    print("读取失败: 等待 1.5 秒未等到新数据")

四、多线程实战：生产者-消费者与 task_done/join 协同

在实际生产环境中，queue.Queue 通常与 task_done() 和 join() 结合使用。消费者线程从队列取出并处理完一个项目后，必须调用一次 q.task_done()；主线程或监控线程调用 q.join() 能够阻塞等待，直到队列中所有已被 put 的任务都被消费并确认完毕。

图 3：生产者与消费者之间通过 task_done 与 join 实现闭环同步

下面是一个可直接执行的完整代码示例，创建 2 个工作线程并发消费 5 个待处理任务：

import queue
import threading
import time

# 创建任务队列，容量限制为 10
work_queue = queue.Queue(maxsize=10)

def worker(worker_id):
    while True:
        try:
            # 阻塞等待任务，若超过 2 秒未收到新任务则退出循环
            task = work_queue.get(block=True, timeout=2.0)
        except queue.Empty:
            print(f"[Worker-{worker_id}] 无新任务，线程准备退出")
            break

        print(f"[Worker-{worker_id}] 正在处理: {task}")
        time.sleep(0.5)  # 模拟耗时任务
        
        # 汇报当前任务处理完毕
        work_queue.task_done()
        print(f"[Worker-{worker_id}] 完成任务: {task}")

# 启动 2 个消费者线程
threads = []
for i in range(1, 3):
    t = threading.Thread(target=worker, args=(i,))
    t.start()
    threads.append(t)

# 生产者向队列写入 5 项工作
for task_id in range(1, 6):
    work_queue.put(f"任务编号_{task_id}")
    print(f"[MainThread] 成功写入 任务编号_{task_id}")

# 等待队列中所有任务执行 task_done()
work_queue.join()
print("[MainThread] 队列中全部任务已处理完毕！")

# 等待工作线程安全终止
for t in threads:
    t.join()
print("[MainThread] 程序平稳退出")

五、常用辅助方法与核心 API 速查

除 put 和 get 之外，queue.Queue 还提供了多个辅助监控与状态查询方法：

图 4：queue.Queue 核心接口与分类功能一览

• q.qsize()：返回队列当前的大致元素数量。请注意：在并发环境中，返回结果只是瞬时快照，不能作为“随后立即读取或写入不会阻塞”的绝对保证。
• q.empty()：如果队列为空返回 True，反之返回 False。同样具有并发时效性限制。
• q.full()：如果队列元素达到 maxsize 则返回 True。

编写高可靠并发程序时，推荐优先使用 try...except queue.Empty / queue.Full 的异常捕获模式，或显式传入 timeout 参数，避免依赖 empty() 或 full() 条件判断可能引发的时序竞争漏洞。