Python 队列示例

from queue import Queue
q = Queue(maxsize=10)  # 限制容量
q.put('task1')
task = q.get()

import threading
from queue import Queue

def worker(q):
    while True:
        item = q.get()

       if item is None:
            break
        process(item)
        q.task_done()

q = Queue()
threads = []
for _ in range(4):
    t = threading.Thread(target=worker, args=(q,))
    t.start()
    threads.append(t)

for item in range(10):
    q.put(item)

q.join()
for _ in threads:
    q.put(None)
for t in threads:
    t.join()

有关线程模式的更多信息，请参见此 Python 线程示例。

多进程队列

multiprocessing 模块提供了自己的 Queue 以实现进程安全的通信。与 queue.Queue 不同，这个队列在底层使用管道和锁。

from multiprocessing import Process, Queue

def worker(q):
    q.put('来自进程的结果')

if __name__ == '__main__':
    q = Queue()
    p = Process(target=worker, args=(q,))
    p.start()
    print(q.get())  # 从子进程接收数据
    p.join()

实用提示：

• 避免使用大型对象；它们可能会减慢序列化速度。
• 完成后在进程上使用 join()，在队列上使用 close()。
• 对于许多生产者，考虑使用 Manager().Queue()。

这可以让你在多个核心之间安全地并行处理 CPU 密集型任务。

优先级队列

当顺序按优先级重要时，切换到 PriorityQueue。每个条目是一个元组 (priority, data)。

from queue import PriorityQueue

pq = PriorityQueue()
pq.put((2, '低优先级任务'))

pq.put((1, '高优先级任务'))
print(pq.get()[1])  # 打印 '高优先级任务'

使用案例：

• 按截止日期调度任务。
• 在网络服务器中管理任务优先级。
• 实时事件处理。

注意：较小的数字优先输出。对于相同优先级的情况，采用先进先出（FIFO）规则进行打破平局。

将优先级与工作线程混合，只需将 Queue 替换为 PriorityQueue 即可。

使用 asyncio 的队列

在异步代码中，使用 asyncio.Queue。它与事件循环集成，并使用 await 而不是阻塞调用。

import asyncio

async def producer(q):
    for i in range(5):
        await q.put(i)
        print(f'put {i}')

async def consumer(q):
    while True:
        item = await q.get()
        print(f'获得 {item}')
        q.task_done()

async def main():
    q = asyncio.Queue()
    prod = asyncio.create_task(producer(q))

cons = asyncio.create_task(consumer(q))
    await prod
    await q.join()
    cons.cancel()

asyncio.run(main())

这种模式使您的异步生产者和消费者在没有线程的情况下保持同步。

常见陷阱和提示

即使有内置的安全性，队列也可能被误用：

• 忘记调用 task_done() 会导致 join() 挂起。
• 没有哨兵值意味着线程永远不会退出。
• 队列过满会意外阻塞生产者。

最佳实践：

• 始终发出线程退出的信号（例如，使用 None 作为哨兵）。
• 选择较小的 maxsize 以防止内存膨胀。
• 如果使用超时，请将 get() 包裹在 try/except 中。

通过及早发现这些问题，您可以在流量激增时保持系统的可靠性。

结论

Python 的队列模块是管理并发的强大助手。从简单的 FIFO 队列到优先级调度和异步工作流，每个类都解决了明确的需求。了解何时使用 Queue、PriorityQueue 或 asyncio.Queue 有助于避免死锁、数据丢失以及线程或进程挂起。通过实际示例练习并遵循上述提示，将使您的代码更加健壮和可维护。现在轮到您了：选择一种队列类型，构建一个小型的生产者-消费者任务，看看并发如何运作。