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

拾光

发布时间：2026-10-29 | 5人浏览过

原创

详细讲解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] 程序平稳退出")

运行提示：task_done() 调用的次数必须严格等于成功 get() 的次数。如果调用次数多于未完成项，Python 会抛出 ValueError('task_done() called too many times')；若少于则会导致 join() 永远无法唤醒并引发死锁。