在 Go 语言中，使用 channels 实现生产者-消费者模式是一种常见的并发编程模式。生产者负责生成数据，消费者负责处理数据。通过 channels，生产者和消费者可以安全地交换数据，无需显式加锁。本文将从基础实现开始，逐步深入到各种工程级实现方案，涵盖无缓冲/有缓冲 channel、多生产者多消费者、优雅关闭、背压处理、超时控制和性能对比等主题。
生产者-消费者模式的核心原理
生产者-消费者模式是一种经典的并发设计模式，用于解耦数据生产与数据处理的速度差异。在 Go 中，这一模式天然适合使用 channel 来实现，因为 channel 本身就提供了 goroutine 间的安全通信和同步机制。
模式的核心角色包括：
- 生产者（Producer）：负责生成数据，将数据发送到 channel
- 消费者（Consumer）：负责从 channel 接收数据并进行处理
- 缓冲通道（Buffer）：在生产者和消费者之间解耦，平衡双方的速度差异
Go 的 channel 设计完美契合这一模式：发送操作在缓冲区满时阻塞，接收操作在缓冲区空时阻塞，这种天然的背压机制使得实现非常优雅。
基础实现：无缓冲 Channel
无缓冲 channel 要求发送和接收同时发生，这是一种强同步的生产者-消费者模型。
package main
import (
"fmt"
"time"
)
func producer(ch chan<- int) {
for i := 1; i <= 5; i++ {
fmt.Printf("Producer: generating %d\n", i)
ch <- i // 发送数据，会阻塞直到消费者接收
}
close(ch) // 生产完毕，关闭 channel
fmt.Println("Producer: done")
}
func consumer(ch <-chan int) {
for v := range ch { // 从 channel 接收数据，channel 关闭后自动退出循环
fmt.Printf("Consumer: processing %d\n", v)
time.Sleep(100 * time.Millisecond) // 模拟处理耗时
}
fmt.Println("Consumer: done")
}
func main() {
ch := make(chan int) // 无缓冲 channel
go producer(ch)
consumer(ch) // 主 goroutine 作为消费者
}
在这个示例中，生产者的每次发送都会等待消费者接收后才能继续，形成了严格的同步关系。这种模式适用于生产者不希望过度领先于消费者的场景。
有缓冲 Channel 实现
有缓冲 channel 允许生产者在没有消费者就绪时也能继续生产，只要缓冲区未满。
package main
import (
"fmt"
"time"
)
func producer(ch chan<- int) {
for i := 1; i <= 10; i++ {
ch <- i
fmt.Printf("Producer: sent %d (buffer may have space)\n", i)
}
close(ch)
}
func consumer(id int, ch <-chan int) {
for v := range ch {
fmt.Printf("Consumer %d: received %d\n", id, v)
time.Sleep(200 * time.Millisecond)
}
}
func main() {
ch := make(chan int, 5) // 缓冲区大小为 5
go producer(ch)
go consumer(1, ch) // 启动两个消费者
go consumer(2, ch)
time.Sleep(3 * time.Second) // 等待处理完成
}
有缓冲 channel 的缓冲区大小是一个重要的调优参数：
- 缓冲区过小：生产者频繁阻塞，无法充分发挥并发优势
- 缓冲区过大：占用更多内存，消费者处理延迟增加
- 一般建议：根据生产速度和消费速度的差值来设置，经验公式是
(生产速率 - 消费速率) * 容忍延迟
带优雅关闭的生产者消费者
在生产环境中，我们需要一种安全、有序的方式来关闭整个生产者-消费者系统。直接关闭 channel 可能导致正在发送的生产者 panic。
package main
import (
"fmt"
"sync"
"time"
)
func producer(id int, ch chan<- int, done <-chan struct{}, wg *sync.WaitGroup) {
defer wg.Done()
i := 0
for {
select {
case <-done:
fmt.Printf("Producer %d: shutting down\n", id)
return
default:
i++
ch <- i
fmt.Printf("Producer %d: sent %d\n", id, i)
time.Sleep(100 * time.Millisecond)
}
}
}
func consumer(id int, ch <-chan int, done <-chan struct{}, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case v, ok := <-ch:
if !ok {
fmt.Printf("Consumer %d: channel closed, exiting\n", id)
return
}
fmt.Printf("Consumer %d: processing %d\n", id, v)
time.Sleep(300 * time.Millisecond)
case <-done:
fmt.Printf("Consumer %d: shutdown signal received\n", id)
return
}
}
}
func main() {
ch := make(chan int, 10)
done := make(chan struct{})
var wg sync.WaitGroup
// 启动 2 个生产者
for i := 0; i < 2; i++ {
wg.Add(1)
go producer(i+1, ch, done, &wg)
}
// 启动 3 个消费者
for i := 0; i < 3; i++ {
wg.Add(1)
go consumer(i+1, ch, done, &wg)
}
// 运行一段时间后优雅关闭
time.Sleep(2 * time.Second)
fmt.Println("Main: initiating shutdown")
close(done) // 发送关闭信号
// 等待所有生产者和消费者退出
wg.Wait()
close(ch) // 安全关闭 channel（此时所有生产者已退出）
fmt.Println("Main: all goroutines stopped")
}
注意这里的关闭顺序非常重要：
- 关闭
donechannel 通知所有 goroutine 停止工作 - 等待生产者退出并关闭数据 channel
- 消费者检测到 channel 关闭后也退出
多生产者多消费者实现
当任务量巨大时，单个生产者和单个消费者往往成为瓶颈。扩展为多个生产者和多个消费者是常见的需求。
package main
import (
"fmt"
"sync"
)
func producer(id int, ch chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for i := 0; i < 5; i++ {
data := id*100 + i
ch <- data
fmt.Printf("Producer %d: sent %d\n", id, data)
}
}
func consumer(id int, ch <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for data := range ch {
fmt.Printf("Consumer %d: received %d\n", id, data)
}
}
func main() {
ch := make(chan int, 20)
var producerWg, consumerWg sync.WaitGroup
// 启动 3 个生产者
for i := 1; i <= 3; i++ {
producerWg.Add(1)
go producer(i, ch, &producerWg)
}
// 启动 2 个消费者
for i := 1; i <= 2; i++ {
consumerWg.Add(1)
go consumer(i, ch, &consumerWg)
}
// 等待所有生产者完成
producerWg.Wait()
close(ch) // 所有生产者完成后关闭 channel
// 等待所有消费者完成
consumerWg.Wait()
fmt.Println("All tasks completed")
}
关键点：使用两个独立的 sync.WaitGroup 分别跟踪生产者和消费者。生产者的 WaitGroup 完成后才关闭 channel，然后消费者的 range 循环自然结束。
背压（Backpressure）处理
背压是指当消费者处理速度跟不上生产者时，系统需要有机制来控制生产速率，防止内存无限增长。
package main
import (
"fmt"
"time"
)
// 带背压的信号 channel
type BackpressureProducer struct {
dataCh chan int
ackCh chan struct{}
batchSize int
}
func NewBackpressureProducer(batchSize int) *BackpressureProducer {
return &BackpressureProducer{
dataCh: make(chan int, batchSize),
ackCh: make(chan struct{}, batchSize),
batchSize: batchSize,
}
}
func (p *BackpressureProducer) Produce(value int) bool {
select {
case p.dataCh <- value:
return true
default:
// 缓冲区满，等待消费者确认
select {
case <-p.ackCh:
p.dataCh <- value
return true
case <-time.After(time.Second):
return false // 超时，放弃生产
}
}
}
func (p *BackpressureProducer) Consume() (int, bool) {
select {
case v := <-p.dataCh:
// 消费后发送确认，让生产者继续
select {
case p.ackCh <- struct{}{}:
default:
}
return v, true
case <-time.After(time.Second):
return 0, false
}
}
func main() {
producer := NewBackpressureProducer(3)
// 生产者 goroutine
go func() {
for i := 1; i <= 10; i++ {
if producer.Produce(i) {
fmt.Printf("Produced: %d\n", i)
} else {
fmt.Printf("Failed to produce: %d (backpressure)\n", i)
}
time.Sleep(50 * time.Millisecond)
}
close(producer.dataCh)
}()
// 消费者
for {
v, ok := producer.Consume()
if !ok {
if len(producer.dataCh) == 0 {
break
}
continue
}
fmt.Printf("Consumed: %d\n", v)
time.Sleep(300 * time.Millisecond) // 慢消费
}
}
更简单的背压实现方式是利用有缓冲 channel 的天然阻塞特性——当缓冲区满时，生产者自动阻塞，这正是最自然的背压机制。
超时与取消机制
在生产环境中，生产者和消费者都需要处理超时和任务取消的情况。
package main
import (
"context"
"fmt"
"time"
)
func producer(ctx context.Context, ch chan<- int) {
i := 0
for {
select {
case <-ctx.Done():
fmt.Println("Producer: context cancelled")
return
default:
i++
select {
case ch <- i:
fmt.Printf("Producer: sent %d\n", i)
case <-ctx.Done():
fmt.Println("Producer: context cancelled while sending")
return
}
time.Sleep(100 * time.Millisecond)
}
}
}
func consumer(ctx context.Context, id int, ch <-chan int) {
for {
select {
case v, ok := <-ch:
if !ok {
fmt.Printf("Consumer %d: channel closed\n", id)
return
}
fmt.Printf("Consumer %d: processing %d\n", id, v)
time.Sleep(200 * time.Millisecond)
case <-ctx.Done():
fmt.Printf("Consumer %d: context cancelled\n", id)
return
}
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
ch := make(chan int, 5)
go producer(ctx, ch)
go consumer(ctx, 1, ch)
go consumer(ctx, 2, ch)
<-ctx.Done()
fmt.Println("Main: timeout reached")
close(ch)
time.Sleep(500 * time.Millisecond) // 等待 goroutine 退出
}
使用 context.Context 是 Go 中管理 goroutine 生命周期的标准做法。它支持超时、取消和值传递，比手动使用 done channel 更加灵活和标准化。
与 sync.WaitGroup 和 context 的结合
将生产者-消费者模式与 sync.WaitGroup 和 context 结合起来，可以构建出健壮的生产级代码：
package main
import (
"context"
"fmt"
"sync"
"time"
)
type Task struct {
ID int
Data string
}
type Result struct {
TaskID int
Output string
Duration time.Duration
}
func produce(ctx context.Context, tasks chan<- Task, wg *sync.WaitGroup) {
defer wg.Done()
for i := 1; i <= 10; i++ {
select {
case <-ctx.Done():
fmt.Println("Producer: cancelled")
return
case tasks <- Task{ID: i, Data: fmt.Sprintf("task-%d", i)}:
fmt.Printf("Produced task %d\n", i)
}
}
}
func consume(ctx context.Context, id int, tasks <-chan Task, results chan<- Result, wg *sync.WaitGroup) {
defer wg.Done()
for {
select {
case <-ctx.Done():
fmt.Printf("Consumer %d: cancelled\n", id)
return
case task, ok := <-tasks:
if !ok {
return
}
start := time.Now()
time.Sleep(100 * time.Millisecond) // 模拟处理
results <- Result{
TaskID: task.ID,
Output: fmt.Sprintf("result-%d", task.ID),
Duration: time.Since(start),
}
}
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()
tasks := make(chan Task, 5)
results := make(chan Result, 10)
var producerWg sync.WaitGroup
producerWg.Add(1)
go produce(ctx, tasks, &producerWg)
var consumerWg sync.WaitGroup
for i := 1; i <= 3; i++ {
consumerWg.Add(1)
go consume(ctx, i, tasks, results, &consumerWg)
}
// 等待生产者完成，然后关闭 tasks channel
go func() {
producerWg.Wait()
close(tasks)
}()
// 等待消费者完成，然后关闭 results channel
go func() {
consumerWg.Wait()
close(results)
}()
// 收集并打印结果
for r := range results {
fmt.Printf("Result: TaskID=%d, Output=%s, Duration=%v\n",
r.TaskID, r.Output, r.Duration)
}
fmt.Println("All done")
}
性能对比：不同 Buffer 大小的影响
channel 的缓冲区大小对性能有显著影响。以下基准测试展示了不同配置下的吞吐量差异：
package main
import (
"fmt"
"sync"
"time"
)
func benchmark(bufferSize int, producerCount, consumerCount int) time.Duration {
ch := make(chan int, bufferSize)
var producerWg, consumerWg sync.WaitGroup
itemCount := 10000
start := time.Now()
// 启动消费者
for i := 0; i < consumerCount; i++ {
consumerWg.Add(1)
go func() {
defer consumerWg.Done()
for range ch {
}
}()
}
// 启动生产者
eachProducerCount := itemCount / producerCount
for i := 0; i < producerCount; i++ {
producerWg.Add(1)
go func() {
defer producerWg.Done()
for j := 0; j < eachProducerCount; j++ {
ch <- j
}
}()
}
producerWg.Wait()
close(ch)
consumerWg.Wait()
return time.Since(start)
}
func main() {
configs := []struct {
bufferSize int
producerCount int
consumerCount int
}{
{0, 1, 1},
{1, 1, 1},
{10, 1, 1},
{100, 1, 1},
{1000, 1, 1},
{100, 2, 2},
{100, 4, 4},
{1000, 4, 8},
}
fmt.Printf("%-10s %-10s %-10s %-15s\n", "Buffer", "Producers", "Consumers", "Duration")
fmt.Println(string(make([]byte, 50)))
for _, c := range configs {
d := benchmark(c.bufferSize, c.producerCount, c.consumerCount)
fmt.Printf("%-10d %-10d %-10d %-15v\n",
c.bufferSize, c.producerCount, c.consumerCount, d)
}
}
一般规律：
- 无缓冲 channel 吞吐量最低，但同步最强
- 随着缓冲区增大，吞吐量提升但边际递减
- 增加生产者和消费者数量通常比单纯增大缓冲区更有效
- 最优配置取决于具体场景，需要通过基准测试来确定
完整可运行代码：带统计和监控的工作队列
以下是一个生产级的生产者-消费者实现，包含任务统计、错误处理、限流和优雅关闭：
package main
import (
"context"
"fmt"
"sync"
"sync/atomic"
"time"
)
// Task 工作单元
type Task struct {
ID int
Payload string
}
// Stats 统计信息
type Stats struct {
Produced int64
Consumed int64
Errors int64
StartTime time.Time
}
func (s *Stats) Throughput() float64 {
elapsed := time.Since(s.StartTime).Seconds()
if elapsed == 0 {
return 0
}
return float64(atomic.LoadInt64(&s.Consumed)) / elapsed
}
// WorkerPool 生产者-消费者工作池
type WorkerPool struct {
tasks chan Task
stats *Stats
workers int
wg sync.WaitGroup
ctx context.Context
cancel context.CancelFunc
}
func NewWorkerPool(bufferSize, workers int) *WorkerPool {
ctx, cancel := context.WithCancel(context.Background())
return &WorkerPool{
tasks: make(chan Task, bufferSize),
stats: &Stats{StartTime: time.Now()},
workers: workers,
ctx: ctx,
cancel: cancel,
}
}
func (wp *WorkerPool) Start() {
// 启动消费者（worker）
for i := 0; i < wp.workers; i++ {
wp.wg.Add(1)
go wp.worker(i + 1)
}
}
func (wp *WorkerPool) worker(id int) {
defer wp.wg.Done()
for {
select {
case <-wp.ctx.Done():
fmt.Printf("Worker %d: stopping\n", id)
return
case task, ok := <-wp.tasks:
if !ok {
fmt.Printf("Worker %d: channel closed, exiting\n", id)
return
}
wp.process(task)
}
}
}
func (wp *WorkerPool) process(task Task) {
// 模拟处理
time.Sleep(10 * time.Millisecond)
// 模拟偶尔出错
if task.ID%10 == 0 {
atomic.AddInt64(&wp.stats.Errors, 1)
fmt.Printf("Task %d: processing error\n", task.ID)
return
}
atomic.AddInt64(&wp.stats.Consumed, 1)
}
func (wp *WorkerPool) Submit(task Task) bool {
select {
case <-wp.ctx.Done():
return false
case wp.tasks <- task:
atomic.AddInt64(&wp.stats.Produced, 1)
return true
case <-time.After(time.Second):
return false // 投递超时（背压）
}
}
func (wp *WorkerPool) Stop() {
wp.cancel()
close(wp.tasks)
wp.wg.Wait()
}
func (wp *WorkerPool) Report() {
fmt.Printf("\n========== Statistics ==========\n")
fmt.Printf("Produced: %d\n", atomic.LoadInt64(&wp.stats.Produced))
fmt.Printf("Consumed: %d\n", atomic.LoadInt64(&wp.stats.Consumed))
fmt.Printf("Errors: %d\n", atomic.LoadInt64(&wp.stats.Errors))
fmt.Printf("Pending: %d\n", len(wp.tasks))
fmt.Printf("Throughput: %.2f items/sec\n", wp.stats.Throughput())
fmt.Printf("================================\n")
}
func main() {
pool := NewWorkerPool(100, 4)
pool.Start()
// 启动生产者
go func() {
for i := 1; i <= 500; i++ {
if !pool.Submit(Task{ID: i, Payload: fmt.Sprintf("data-%d", i)}) {
fmt.Printf("Failed to submit task %d\n", i)
}
}
}()
// 运行一段时间后优雅关闭
time.Sleep(2 * time.Second)
fmt.Println("Main: initiating shutdown")
pool.Stop()
pool.Report()
}
完整可运行代码：限流版生产者-消费者
当消费者处理能力有限时，使用令牌桶算法进行限流：
package main
import (
"context"
"fmt"
"sync"
"time"
)
// TokenBucket 令牌桶限流器
type TokenBucket struct {
tokens chan struct{}
interval time.Duration
stop chan struct{}
}
func NewTokenBucket(rate int, interval time.Duration) *TokenBucket {
tb := &TokenBucket{
tokens: make(chan struct{}, rate),
interval: interval,
stop: make(chan struct{}),
}
go tb.fill(rate)
return tb
}
func (tb *TokenBucket) fill(rate int) {
ticker := time.NewTicker(tb.interval)
defer ticker.Stop()
for {
select {
case <-ticker.C:
for i := 0; i < rate; i++ {
select {
case tb.tokens <- struct{}{}:
default:
}
}
case <-tb.stop:
return
}
}
}
func (tb *TokenBucket) Acquire(ctx context.Context) bool {
select {
case <-tb.tokens:
return true
case <-ctx.Done():
return false
}
}
func (tb *TokenBucket) Stop() {
close(tb.stop)
}
func main() {
// 每秒产生 10 个令牌
bucket := NewTokenBucket(10, time.Second)
defer bucket.Stop()
tasks := make(chan int, 20)
var wg sync.WaitGroup
// 生产者（高速生产）
go func() {
for i := 1; i <= 50; i++ {
tasks <- i
fmt.Printf("Produced: %d\n", i)
}
close(tasks)
}()
// 消费者（受令牌桶限流）
for i := 0; i < 3; i++ {
wg.Add(1)
go func(id int) {
defer wg.Done()
for task := range tasks {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
if bucket.Acquire(ctx) {
fmt.Printf("Consumer %d: processing %d (rate limited)\n", id, task)
time.Sleep(50 * time.Millisecond)
} else {
fmt.Printf("Consumer %d: task %d dropped (no token)\n", id, task)
}
cancel()
}
}(i + 1)
}
wg.Wait()
fmt.Println("All tasks processed with rate limiting")
}
常见错误与死锁排查
错误一：在消费者之前关闭 channel
如果在消费者还在运行时由消费者方关闭 channel，会导致向已关闭 channel 发送数据时 panic。channel 应该由发送方（生产者）关闭。
错误二：忘记关闭 channel
如果生产者不关闭 channel，消费者的 range 循环将永远阻塞，导致 goroutine 泄漏。
错误三：多生产者中某个生产者关闭 channel
多个生产者共享一个 channel 时，不能由某个生产者单独关闭。应该使用 sync.WaitGroup 等待所有生产者完成后再关闭。
错误四：select 中 nil channel
向 nil channel 发送或接收会永久阻塞。确保所有在 select 中使用的 channel 都经过初始化。
死锁排查方法
- 使用
go run -race或go test -race检测竞态 - 使用
go tool pprof查看 goroutine 堆栈 - 启用 GODEBUG 环境变量获取更多调度信息
- 在关键位置添加超时逻辑，避免永久阻塞
总结
生产者-消费者模式是 Go 并发编程中最基础也最实用的模式之一。Go 的 channel 设计让这一模式的实现变得异常简洁优雅。从最简单的无缓冲 channel 到复杂的带限流、背压、超时和优雅关闭的工程实现，你可以根据具体场景选择合适的方案。
记住几个核心原则：
- channel 由发送方关闭，接收方检查是否已关闭
- 使用
sync.WaitGroup协调多个 goroutine 的生命周期 - 使用
context.Context处理超时和取消 - 利用有缓冲 channel 的天然阻塞特性实现背压
- 通过基准测试找到最优的缓冲区大小和 worker 数量
掌握了这些，你就能在生产环境中稳定、高效地使用生产者-消费者模式解决各种并发问题。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。