如何使用Golang实现生产者消费者模型_Golang channel与goroutine实践

直接用无缓冲channel易卡死，因发送/接收未配对时goroutine阻塞；需用sync.WaitGroup协调关闭时机，或引入context实现可控生命周期与错误传播。

如何使用Golang实现生产者消费者模型_Golang channel与goroutine实践第1张

为什么直接用 `chan` 做生产者消费者容易卡死

因为无缓冲 channel 在发送和接收未配对时会阻塞 goroutine，一旦生产者发完、消费者还没启动，或消费者处理太慢，整个流程就停在 send 或 recv 上。常见现象是程序启动后没输出、CPU 占用低、goroutine 数卡在非 0 值但无进展。

关键不是“能不能用 channel”，而是怎么控制生命周期和背压。必须显式管理：关闭 channel 的时机、消费者退出条件、是否允许丢弃数据。

不要在生产者 goroutine 里直接 close(ch) 后还往 ch 发数据（panic: send on closed channel）
不要让多个生产者共用一个未加锁的 sync.WaitGroup 计数器
如果消费者 panic 未 recover，且没有超时或 context 控制，整个管道会不可恢复挂起

用 `sync.WaitGroup` + `close()` 实现基础可靠模型

这是最常被抄错的写法：很多人把 close(ch) 放在生产者末尾，却没等所有生产者完成；或消费者用 for range ch 但生产者还没关 channel 就提前退出。

正确做法是：用 sync.WaitGroup 等待所有生产者结束，再统一 close；消费者只在 channel 关闭后自然退出。

立即学习“go语言免费学习笔记（深入）”；

func main() {
    ch := make(chan int, 10)
    var wg sync.WaitGroup
// 启动 2 个生产者
for i := 0; i < 2; i++ {
    wg.Add(1)
    go func(id int) {
        defer wg.Done()
        for j := 0; j < 5; j++ {
            ch <- id*10 + j
        }
    }(i)
}

// 启动 1 个消费者
go func() {
    wg.Wait() // 等所有生产者完成
    close(ch) // 此时才安全关闭
}()

// 消费者用 for-range，自动在 close 后退出
for val := range ch {
    fmt.Println("consumed:", val)
}
}
带 context 控制和错误传播的工业级写法
真实场景中，你不能靠“等所有生产者发完”来关 channel —— 生产者可能因网络超时、数据库错误提前失败，或需要支持热停止。这时必须引入 context.Context。

							
核心原则：channel 关闭由单一 goroutine 主导，且该 goroutine 必须监听 ctx.Done()。消费者也需检查 ctx.Err() 并及时退出，避免泄漏。


select 语句里必须同时监听 ch 和 ctx.Done()，不能只读 channel
生产者若遇到错误，应调用 cancel()，而不是自己 close channel
用 errgroup.Group 可简化多 goroutine 错误收集，但要注意它默认不支持取消信号透传
func runPipeline(ctx context.Context, ch chan int) error {
    g, ctx := errgroup.WithContext(ctx)
// 生产者
g.Go(func() error {
    for i := 0; i < 10; i++ {
        select {
        case ch <- i:
        case <-ctx.Done():
            return ctx.Err()
        }
    }
    return nil
})

// 消费者
g.Go(func() error {
    for {
        select {
        case val, ok := <-ch:
            if !ok {
                return nil
            }
            fmt.Println("got:", val)
        case <-ctx.Done():
            return ctx.Err()
        }
    }
})

return g.Wait()
}
缓冲区大小设多少？别硬编码 make(chan int, 100)

缓冲区不是越大越好。设太大浪费内存（尤其 channel 存指针或大结构体），设太小又频繁阻塞，掩盖真实吞吐瓶颈。实际值取决于：单次生产耗时、单次消费耗时、峰值流量倍数。
更稳妥的做法是：先用无缓冲 channel 测出 baseline 吞吐，再按「平均生产间隔 × 消费延迟 × 安全系数（1.5~3）」估算缓冲大小。线上服务建议配合 metrics 上报 len(ch)，动态调整。

HTTP 请求入队场景：缓冲区 = QPS × P99 处理延迟 × 2
日志采集：用 ring buffer 替代 channel（如 github.com/cespare/xxhash 配合 slice）更省内存
如果消费者可能长时间阻塞（如调外部 API），缓冲区无法解决根本问题，得加限流（golang.org/x/time/rate）或降级逻辑

真正难的从来不是“怎么写出来”，而是当并发量翻 5 倍、下游响应变慢 10 倍、日志里突然出现大量 send on closed channel 时，你能否从 goroutine stack 和 channel 状态里快速定位是哪个环节没遵循关闭契约。