怎么实现一个生产者消费者模式的例子？

wen java案例 2026-06-06 79

生产者-消费者模式从原理到实战（附Go/Python代码）

目录导读

什么是生产者-消费者模式？ – 核心概念与动机
为什么需要这个模式？ – 解决的核心问题
实现该模式的关键要素 – 缓冲区、互斥锁、条件变量
经典实现方式对比 – 阻塞队列、信号量、Channel
Go语言实战：用Channel轻松实现 – 完整代码与运行效果
Python实战：用Queue + threading实现 – 代码与异常处理
常见问题与性能调优 – 死锁预防、容量规划、多消费者
问答环节 – 面试高频问题与深度解答

什么是生产者-消费者模式？

生产者-消费者模式是一种经典的多线程/协程协作设计模式，它将“生成数据”的任务（生产者）与“处理数据”的任务（消费者）解耦，中间通过一个共享的缓冲区（如队列）进行通信。

怎么实现一个生产者消费者模式的例子？

形象类比：一家餐厅，厨师（生产者）做好菜放入传菜台（缓冲区），服务员（消费者）取菜送给客人，两者不需要直接沟通，而是通过传菜台协调节奏。

为什么需要这个模式？

解耦：生产者和消费者不需要知道对方的存在，修改一方不影响另一方。
支持并发：生产者可以持续生产而不用等待消费者处理完毕；消费者也可以持续消费，无需等待生产者。
削峰填谷：当生产速度瞬间超过消费速度时，缓冲区能临时存储数据，避免系统崩溃。
负载均衡：可以灵活增加消费者数量来提升处理能力。

实现该模式的关键要素

要素	作用	常见实现方式
缓冲区	存储临时数据	数组、链表、队列
互斥锁	保护缓冲区并发访问	`sync.Mutex`、`threading.Lock`
条件变量/信号量	通知生产/消费时机	`sync.Cond`、`Semaphore`
结束标志	优雅停止协作	`bool` 标志或特殊终止元素

缺少任何一个要素,都可能导致死锁或数据竞争。

经典实现方式对比

方案	语言/库	特点	适用场景
阻塞队列	Java `BlockingQueue`	自动阻塞，无锁实现（CAS）	高并发Java应用
信号量	C/POSIX	灵活但易出错	操作系统底层
Channel	Go	语言原生支持，自动同步	Go并发编程
Queue + Event	Python	标准库稳定，调试方便	中小型任务队列

Go语言实战：用Channel轻松实现

Go的chan天然支持生产者-消费者模式，简洁且安全。

代码示例（完整可运行）

package main
import (
    "fmt"
    "sync"
    "time"
)
func main() {
    const bufferSize = 5
    jobs := make(chan int, bufferSize)  // 缓冲区
    var wg sync.WaitGroup
    // 生产者
    wg.Add(1)
    go func() {
        defer wg.Done()
        for i := 1; i <= 10; i++ {
            fmt.Printf("生产者: 生产 %d\n", i)
            jobs <- i
            time.Sleep(100 * time.Millisecond)
        }
        close(jobs)  // 关键：关闭通道通知消费者结束
        fmt.Println("生产者: 完成生产")
    }()
    // 消费者
    wg.Add(1)
    go func() {
        defer wg.Done()
        for job := range jobs {  // 通道关闭后自动退出循环
            fmt.Printf("消费者: 消费 %d\n", job)
            time.Sleep(200 * time.Millisecond) // 模拟处理耗时
        }
        fmt.Println("消费者: 完成所有消费")
    }()
    wg.Wait()
    fmt.Println("主程序结束")
}

运行效果示例

生产者: 生产 1
消费者: 消费 1
生产者: 生产 2
生产者: 生产 3
消费者: 消费 2
生产者: 生产 4
... （消费者慢于生产者，缓冲区填满后生产者自动阻塞）
生产者: 完成生产
消费者: 消费 9
消费者: 消费 10
消费者: 完成所有消费
主程序结束

关键点：

带缓冲的chan相当于一个有容量上限的队列。
close(jobs)后消费者range自动退出，避免死循环。
不需要手动加锁,chan内部实现了同步。

Python实战：用Queue + threading实现

Python的queue.Queue是线程安全的，配合threading.Event控制结束。

代码示例

import threading
import queue
import time
import random
STOP_EVENT = threading.Event()  # 用于通知消费者结束
def producer(q, stop_event):
    for i in range(8):
        if stop_event.is_set():
            break
        item = random.randint(1, 100)
        print(f"生产者: 生成 {item}")
        q.put(item)
        time.sleep(0.1)
    # 放入哨兵元素，通知消费者结束
    q.put(None)
    stop_event.set()
    print("生产者: 完成")
def consumer(q, stop_event):
    while True:
        try:
            item = q.get(timeout=1)  # 1秒超时防止无限阻塞
        except queue.Empty:
            if stop_event.is_set():
                break
            continue
        if item is None:  # 检查哨兵
            break
        print(f"消费者: 处理 {item}")
        time.sleep(0.2)
    print("消费者: 完成")
if __name__ == "__main__":
    q = queue.Queue(maxsize=3)  # 缓冲区容量3
    t1 = threading.Thread(target=producer, args=(q, STOP_EVENT))
    t2 = threading.Thread(target=consumer, args=(q, STOP_EVENT))
    t1.start()
    t2.start()
    t1.join()
    t2.join()
    print("主线程结束")