Python queue 模块
FIFO/LIFO/priority queues, blocking semantics, and the task_done protocol.
Key Points`queue.Queue` is already thread-safe — the usual bug is forgetting `task_done()`, which makes `join()` hang forever.
Three Queue Types
| Class | Order | Use for |
|---|---|---|
| Queue | FIFO | Task pipeline |
| LifoQueue | LIFO | Undo stack, DFS |
| PriorityQueue | Lowest value first | Scheduling by priority |
Python
import queue
q = queue.Queue(maxsize=3)
q.put("a"); q.put("b")
print(q.get(), q.qsize())
pq = queue.PriorityQueue()
pq.put((2, "low"))
pq.put((1, "high"))
print(pq.get())Output
a 1
(1, 'high')Producer / Consumer
get() blocks when empty; get_nowait() raises Empty. Consumers must call task_done() or join() never returns.
Python
import queue
import threading
q = queue.Queue()
def worker():
while True:
item = q.get()
if item is None:
q.task_done()
break
print("handled", item)
q.task_done()
t = threading.Thread(target=worker, daemon=True)
t.start()
for i in range(3):
q.put(i)
q.put(None) # 结束哨兵
q.join()
print("all done")Output
handled 0
handled 1
handled 2
all doneAlways set
maxsize — an unbounded queue eats all memory when consumers lag. A bound gives you backpressure for free.三种队列
| 类 | 顺序 | 适用 |
|---|---|---|
| Queue | 先进先出 | 任务流水线 |
| LifoQueue | 后进先出 | 撤销栈、深度优先 |
| PriorityQueue | 值最小优先 | 按优先级调度 |
Python
import queue
q = queue.Queue(maxsize=3)
q.put("a"); q.put("b")
print(q.get(), q.qsize())
pq = queue.PriorityQueue()
pq.put((2, "low"))
pq.put((1, "high"))
print(pq.get())Output
a 1
(1, 'high')生产者消费者
get() 默认阻塞，队列空时会等；get_nowait() 立刻抛 Empty。消费者处理完必须调 task_done()，主线程的 join() 才会返回。
Python
import queue
import threading
q = queue.Queue()
def worker():
while True:
item = q.get()
if item is None:
q.task_done()
break
print("handled", item)
q.task_done()
t = threading.Thread(target=worker, daemon=True)
t.start()
for i in range(3):
q.put(i)
q.put(None) # 结束哨兵
q.join()
print("all done")Output
handled 0
handled 1
handled 2
all donemaxsize 不要留空：无界队列在消费跟不上时会吃光内存。给个上限，让生产者在 put() 处自然背压。