import threading

import queue

import time

import random

# 队列大小

BUFFER_SIZE = 10

# 队列

buffer = queue.Queue(BUFFER_SIZE)

# 生产者线程

class Producer(threading.Thread):

    def __init__(self, name, queue):

        super().__init__()

        self.name = name

        self.queue = queue

    def run(self):

        for i in range(20):  # 每个生产者生产20个数据

            item = f"{self.name}-item-{i}"

            self.queue.put(item)  # 阻塞直到队列有空间

            print(f"[生产者 {self.name}] 生产了 {item}, 队列大小: {self.queue.qsize()}")

            time.sleep(random.uniform(0.1, 0.5))  # 模拟生产耗时

# 消费者线程

class Consumer(threading.Thread):

    def __init__(self, name, queue):

        super().__init__()

        self.name = name

        self.queue = queue

    def run(self):

        while True:

            item = self.queue.get()  # 阻塞直到队列有数据

            if item is None:

                # 遇到 None 代表生产者完成，退出

                print(f"[消费者 {self.name}] 收到退出信号，结束消费")

                break

            print(f"[消费者 {self.name}] 消费了 {item}, 队列大小: {self.queue.qsize()}")

            time.sleep(random.uniform(0.2, 0.6))  # 模拟消费耗时

            self.queue.task_done()  # 标记任务完成

def main():

    producers = [Producer(f"P{i+1}", buffer) for i in range(2)]  # 两个生产者

    consumers = [Consumer(f"C{i+1}", buffer) for i in range(3)]  # 三个消费者

    # 启动消费者

    for c in consumers:

        c.start()

    # 启动生产者

    for p in producers:

        p.start()

    # 等待生产者结束

    for p in producers:

        p.join()

    # 通知消费者退出（发送 None）

    for _ in consumers:

        buffer.put(None)

    # 等待消费者结束

    for c in consumers:

        c.join()

    print("所有生产者和消费者已完成任务。")

if __name__ == "__main__":

    main()

说明

线程安全队列

使用 queue.Queue 可以自动处理锁问题，不需要手动加锁。

退出机制

消费者线程在队列收到 None 时退出。

多生产者多消费者

可以自由调整生产者和消费者数量。

任务完成标记

queue.task_done() 可以用来跟踪任务是否完成，结合 queue.join() 可以等所有任务完成。