在Go语言中，channel是goroutine之间通信的主要方式，它把数据发送和接收内建为语言级别的操作，天然支持同步与异步两种模式。生产者消费者问题正好可以用channel来解耦数据产生方与数据处理方，生产者只负责向channel写入，消费者只负责从channel读取，双方不直接持有对方引用，执行节奏完全由channel的容量和阻塞规则控制。这样写出来的并发代码更容易理解，也更容易做扩展。
接下来的内容会从最基础的单个生产者单个消费者开始，逐步过渡到有缓冲、多生产者多消费者，并给出优雅退出方案。所有代码都可以在本地Go环境中运行，建议配合go run命令观察输出顺序。
一、用无缓冲channel搭建基础框架
无缓冲channel在发送和接收时必须同时准备好，否则会阻塞。这个特性让生产者与消费者在每一步都保持同步。例如生产者每次发送一个数据，都要等消费者完成接收后才能继续下一次发送，消费者也必须在生产者准备好数据后才能真正拿到值。这种强同步关系可以保证数据不会积压，每一份数据都能被及时处理。
下面是一个最简单的实现：
package main
import (
"fmt"
"time"
)
func producer(ch chan int) {
for i := 1; i <= 5; i++ {
fmt.Printf("生产者发送：%d\n", i)
ch <- i // 无缓冲，必须等消费者接收
}
close(ch)
}
func consumer(ch chan int) {
for v := range ch {
fmt.Printf("消费者接收：%d\n", v)
time.Sleep(100 * time.Millisecond)
}
}
func main() {
ch := make(chan int)
go producer(ch)
consumer(ch)
}
上面的代码中，生产者发送一个数字后必须等待消费者接收才能继续下一个循环。消费者使用range从channel中读取，直到channel被关闭。生产者在发送完所有数据后调用close(ch)来通知消费者没有更多数据。注意，关闭channel必须由生产者完成，因为消费者只会读取，读取已关闭channel会得到零值并在读完缓冲后结束range。如果消费者也去关闭channel，可能出现向已关闭channel写入的panic。
无缓冲channel的同步特性使得代码逻辑非常直观，但吞吐量受限于每一步的同步开销。如果生产者的生产速度远快于消费者的处理速度，那么生产者会频繁阻塞，整体效率不高。此时可以引入有缓冲channel来缓解这一问题。
二、有缓冲channel与容量选择
有缓冲channel允许生产者在消费者尚未准备好时先把数据放入缓冲区，只有缓冲区满时才会阻塞发送。这样可以平滑瞬时流量，提高整体吞吐量。比如把make(chan int)改成make(chan int, 3)，生产者可以连续发送三个数据再被阻塞，消费者可以按自己的节奏从缓冲区取数据。
下面演示容量为5的有缓冲channel：
package main
import (
"fmt"
"time"
)
func producer(ch chan int) {
for i := 1; i <= 10; i++ {
ch <- i
fmt.Printf("生产者已发送：%d\n", i)
}
close(ch)
}
func consumer(ch chan int) {
for v := range ch {
fmt.Printf("消费者处理：%d\n", v)
time.Sleep(200 * time.Millisecond)
}
}
func main() {
ch := make(chan int, 5) // 容量为5
go producer(ch)
consumer(ch)
}
容量为5时，生产者在向channel写入5个数据后，第6次写入会被阻塞，直到消费者取走一个数据腾出空间。输出结果中生产者可能先连续打印多条“已发送”，然后才出现消费者处理，这正是有缓冲channel带来的异步效果。相比无缓冲版本，生产者的阻塞频率明显降低。
选择缓冲容量并没有统一公式，一般需要根据生产速率、消费速率以及可接受的延迟来决定。如果容量太小，接近无缓冲行为；容量太大，虽然写入很少阻塞，但会占用更多内存，并且消费者崩溃时缓冲区中的数据可能来不及处理。在实践里，可以先从较小的值开始，用压测观察goroutine的阻塞情况和内存占用，再逐步调整。有缓冲channel特别适合生产与消费速率不稳定的场景，比如日志采集、任务队列等。
需要特别说明，缓冲channel并不能解决消费者处理不过来的根本问题，它只是提供了一个临时存储。如果长时间消费速率低于生产速率，缓冲区最终会满，生产者仍然会阻塞。真正要改善这种情况，需要考虑增加消费者数量，也就是下一节讨论的多消费者模式。
三、多生产者多消费者与WaitGroup协同
当生产任务可以被并发处理时，单个消费者容易成为瓶颈。可以启动多个消费者goroutine共同从同一个channel读取数据，每个数据只会被一个消费者取走，channel内部保证了并发读取的安全性。同理，多个生产者也可以向同一个channel写入，只要写入互不冲突。下面是一个包含3个生产者和2个消费者的例子：
package main
import (
"fmt"
"sync"
"time"
)
func producer(id int, ch chan<- int, wg *sync.WaitGroup) {
defer wg.Done()
for i := 1; i <= 3; i++ {
ch <- id*10 + i
fmt.Printf("生产者%d发送：%d\n", id, id*10+i)
time.Sleep(50 * time.Millisecond)
}
}
func consumer(id int, ch <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for v := range ch {
fmt.Printf("消费者%d处理：%d\n", id, v)
time.Sleep(100 * time.Millisecond)
}
}
func main() {
ch := make(chan int, 10)
var prodWg sync.WaitGroup
var consWg sync.WaitGroup
// 启动3个生产者
for i := 1; i <= 3; i++ {
prodWg.Add(1)
go producer(i, ch, &prodWg)
}
// 启动2个消费者
for i := 1; i <= 2; i++ {
consWg.Add(1)
go consumer(i, ch, &consWg)
}
// 等待所有生产者完成后关闭channel
prodWg.Wait()
close(ch)
// 等待所有消费者退出
consWg.Wait()
fmt.Println("所有任务处理完成")
}
生产者通过sync.WaitGroup计数，主goroutine等待所有生产者结束后关闭channel。消费者使用range读取，channel关闭后range自动结束，消费者goroutine退出。这里消费者数量比生产者少，但每个数据只会被一个消费者处理。如果把消费者数量增加到4个，吞吐量会进一步提升。
多生产者多消费者的一个关键问题是关闭channel的时机。必须确保所有生产者都已经停止写入后再关闭channel，否则会出现向已关闭channel写入导致的panic。上面的代码用prodWg.Wait()来等待所有生产者完成，之后才执行close(ch)，这在逻辑上是安全的。但消费者也可能因为业务错误需要提前退出，此时需要另一种机制，比如使用context或专门的退出信号，不能简单让消费者也关闭channel。
另外，当channel关闭后，消费者依然可以继续读取缓冲区中剩余的数据，直到读完才会退出range。也就是说close(ch)相当于在数据流末尾加了一个结束标记，并不会丢弃已写入的数据。这一点在使用有缓冲channel时尤为有用。
四、优雅退出与select超时控制
生产环境中，不能让goroutine无限阻塞等待。例如消费者从channel读取数据时，如果生产者长时间没有数据，消费者会一直阻塞。可以使用select结合time.After实现超时，或者通过context取消来通知所有goroutine退出。下面用context.WithTimeout实现带超时的生产者消费者：
package main
import (
"context"
"fmt"
"time"
)
func producer(ctx context.Context, ch chan<- int) {
i := 0
for {
select {
case <-ctx.Done():
fmt.Println("生产者收到退出信号")
close(ch)
return
case ch <- i:
fmt.Printf("生产者发送：%d\n", i)
i++
time.Sleep(100 * time.Millisecond)
}
}
}
func consumer(ctx context.Context, ch <-chan int) {
for {
select {
case <-ctx.Done():
fmt.Println("消费者收到退出信号")
return
case v, ok := <-ch:
if !ok {
fmt.Println("channel已关闭，消费者退出")
return
}
fmt.Printf("消费者处理：%d\n", v)
time.Sleep(200 * time.Millisecond)
}
}
}
func main() {
ch := make(chan int, 5)
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
go producer(ctx, ch)
go consumer(ctx, ch)
<-ctx.Done()
fmt.Println("主程序退出")
time.Sleep(500 * time.Millisecond) // 等待goroutine清理
}
context.WithTimeout设置了2秒的超时，时间一到ctx.Done()会收到信号，生产者和消费者的select都会响应。生产者在退出前关闭channel，消费者检测到channel关闭后退出。这里注意关闭channel是在生产者goroutine内部进行的，保证了不会向已关闭channel写入。
select语句还可以用来实现超时读取，比如消费者不想一直等待生产者，可以设置一个超时分支：
select {
case v := <-ch:
fmt.Println("收到", v)
case <-time.After(500 * time.Millisecond):
fmt.Println("等待超时")
}
这样即使channel没有数据，消费者也能在500毫秒后执行其他逻辑，例如记录日志、上报状态或者重新检查退出条件。在实际项目中，超时控制至关重要，可以避免goroutine泄漏。
另一个常见陷阱是重复关闭channel或者向已关闭channel发送数据。channel关闭后，任何发送操作都会引发panic。为了避免这个问题，通常约定由唯一的所有者（一般是生产者或专门的协调者）负责关闭channel。如果需要多个生产者共同决定何时关闭，可以使用sync.Once来保证close只执行一次，或者在发送前用recover捕获panic，但后者不推荐，因为掩盖了真实错误。更好的做法是把关闭逻辑交给一个单独的goroutine，通过额外的done channel来协调。
掌握这些实践之后，就能根据业务场景灵活选择无缓冲、有缓冲、多消费者以及带超时控制的模式，构建出稳定可靠的Golang并发程序。
Golang channel生产者消费者模式并发编程修改时间：2026-10-05 01:48:09