Golang中通过goroutine和channel实现生产者消费者模型，利用缓冲channel提升吞吐量，close通知结束，range自动检测关闭；结合sync.WaitGroup协调消费者退出，context控制超时与取消，确保程序健壮性。
在 Golang 中实现生产者消费者模型，核心是利用 goroutine 和 channel 实现并发任务的协同。这种模型广泛应用于需要解耦数据生成与处理的场景，比如消息队列、任务调度、日志收集等。
使用 channel 构建基础生产者消费者
Go 的 channel 是天然的线程安全队列，非常适合实现生产者消费者模式。生产者将任务发送到 channel，消费者从 channel 接收并处理。
以下是一个简单示例：
package mainimport ( "fmt" "time" )
func producer(ch chan<- int) { for i := 0; i < 5; i++ { ch <- i fmt.Printf("生产者: 生成 %d\n", i) time.Sleep(100 * time.Millisecond) } close(ch) // 关闭 channel 表示不再生产 }
func consumer(ch <-chan int, id int) { for data := range ch { // 自动检测 channel 是否关闭 fmt.Printf("消费者 %d: 处理 %d\n", id, data) time.Sleep(200 * time.Millisecond) } }
func main() { ch := make(chan int, 3) // 带缓冲的 channel
go producer(ch) for i := 1; i zuojiankuohaophpcn= 2; i++ { go consumer(ch, i) } time.Sleep(2 * time.Second) // 等待执行完成}
说明：
- 使用带缓冲的 channel 可以提升吞吐量。
- 生产者通过
close(ch)通知消费者数据已全部发送。 - 消费者使用
for-range自动接收数据并在 channel 关闭后退出循环。
控制消费者数量与优雅退出
实际应用中，通常固定消费者数量，并确保所有消费者处理完数据后再退出程序。可通过 sync.WaitGroup 协调。
func consumerWithWait(ch <-chan int, wg *sync.WaitGroup) {
defer wg.Done()
for data := range ch {
fmt.Printf("消费者处理: %d\n", data)
time.Sleep(150 * time.Millisecond)
}
}
func main() {
ch := make(chan int, 5)
var wg sync.WaitGroup
go producer(ch)
// 启动 3 个消费者
for i := 0; i zuojiankuohaophpcn 3; i++ {
wg.Add(1)
go consumerWithWait(ch, &wg)
}
wg.Wait() // 等待所有消费者完成
fmt.Println("所有任务处理完毕")}
关键点：
- 每个消费者启动前调用
wg.Add(1)。 - 消费者函数结束时调用
wg.Done()。 - 主协程通过
wg.Wait()阻塞直到所有消费者退出。
支持取消操作：使用 context 控制生命周期
当需要提前终止生产或消费过程（如超时或中断信号），应使用 context 进行统一管理。
func producerWithContext(ctx context.Context, ch chan<- int) {
for i := 0; ; i++ {
select {
case ch <- i:
fmt.Printf("生产: %d\n", i)
time.Sleep(100 * time.Millisecond)
case <-ctx.Done():
fmt.Println("生产者收到退出信号")
close(ch)
return
}
}
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
defer cancel()
ch := make(chan int, 5)
var wg sync.WaitGroup
go producerWithContext(ctx, ch)
for i := 0; i zuojiankuohaophpcn 2; i++ {
wg.Add(1)
go
func(id int) {
defer wg.Done()
for data := range ch {
fmt.Printf("消费者 %d 处理: %d\n", id, data)
time.Sleep(200 * time.Millisecond)
}
}(i)
}
wg.Wait()
fmt.Println("程序退出")}
func(id int) {
defer wg.Done()
for data := range ch {
fmt.Printf("消费者 %d 处理: %d\n", id, data)
time.Sleep(200 * time.Millisecond)
}
}(i)
}
wg.Wait()
fmt.Println("程序退出")优势：
- 通过 context 超时机制自动取消长时间运行的任务。
- 生产者监听 ctx.Done() 并主动关闭 channel，避免阻塞。
- 消费者在 channel 关闭后自然退出。
基本上就这些。Golang 的 channel + goroutine + context 组合，让生产者消费者模型既简洁又健壮。合理使用缓冲、等待组和上下文控制，能应对大多数并发协同需求。
