Go 中多生产者/多消费者模型的并发安全实践

心靈之曲

发布时间：2026-02-05 10:30:10

409人浏览过

来源于php中文网

原创

Go 中多生产者/多消费者模型的并发安全实践

本文详解 go 语言中多生产者-多消费者场景下的典型竞态问题，通过分析原始代码的数据竞争根源，演示如何用原子操作、通道同步与正确关闭机制构建线程安全的并发流水线。

在 Go 的并发模型中，“多个 goroutine 同时修改全局变量”是引发不可预测行为的高发场景。原始代码看似能稳定输出 1–1000 的递增序列，实则掩盖了一个严重问题：对 seq 的非原子读-改-写操作（seq = seq + 1）构成数据竞争（data race）。该竞态未暴露，并非逻辑正确，而是受限于单线程调度（GOMAXPROCS=1）、通道阻塞带来的“伪顺序性”，以及运行环境的偶然性。

? 为什么没看到重复数字？——误解的根源

requestChan 是无缓冲通道，所有消费者发送请求时会阻塞，直到某个 generateStuff goroutine 从该通道接收。这形成了天然的“请求串行化”假象，但不保护后续的 seq 修改。
多个 generateStuff goroutine 可能几乎同时执行
fmt.Println 非并发安全（尤其在 Playground 等受限环境），其输出交织可能进一步掩盖竞态表现；而 log.Printf 内部加锁，更适合调试并发行为。

✅ 正确实现：消除竞态 + 明确生命周期

核心改进点如下：

用 atomic.AddUint64 替代 seq++：保证自增操作的原子性，杜绝计数错误；
使用 range 读取通道 + 显式 close()：避免 goroutine 永久阻塞，实现优雅退出；
移除全局变量依赖，强化通道语义：requestChan 承载“任务分发”，generatorChan 承载“结果返回”，职责清晰；
启用竞态检测器验证：go run -race main.go 是开发阶段必做步骤。

以下是重构后的生产就绪示例：

Cardify卡片工坊

使用Markdown一键生成精美的小红书知识卡片

下载

package main

import (
    "log"
    "sync"
    "sync/atomic"
    "time"
)

var (
    seq        uint64
    generatorChan = make(chan uint64, 10) // 建议缓冲，防结果端阻塞
    requestChan   = make(chan uint64, 10) // 同样缓冲，提升吞吐
)

func generator(genID int) {
    defer func() {
        if r := recover(); r != nil {
            log.Printf("Generator %d panicked: %v", genID, r)
        }
    }()
    for reqID := range requestChan { // 自动退出当 channel 关闭
        s := atomic.AddUint64(&seq, 1) // ✅ 原子递增
        log.Printf("[Gen %2d] Request %3d → Seq %d", genID, reqID, s)
        select {
        case generatorChan <- s:
        case <-time.After(5 * time.Second): // 防死锁兜底
            log.Printf("[Gen %2d] Timeout sending result for req %d", genID, reqID)
        }
    }
}

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()
    for i := 0; i < 5; i++ {
        select {
        case requestChan <- uint64(id):
            result := <-generatorChan
            log.Printf("[Worker %3d] Got result: %4d", id, result)
        case <-time.After(5 * time.Second):
            log.Printf("[Worker %3d] Timeout waiting for generator", id)
            return
        }
    }
}

func main() {
    log.SetFlags(log.Lmicroseconds | log.Lshortfile)

    const (
        numGenerators = 20
        numWorkers    = 200
    )

    var wg sync.WaitGroup

    // 启动生成器
    for i := 0; i < numGenerators; i++ {
        go generator(i)
    }

    // 启动工作者
    wg.Add(numWorkers)
    for i := 0; i < numWorkers; i++ {
        go worker(i, &wg)
    }

    // 等待所有工作者完成
    wg.Wait()

    // 关闭请求通道，通知生成器退出
    close(requestChan)

    // 可选：等待 generatorChan 清空（若需确保所有结果被消费）
    // close(generatorChan) // 若消费者也 range 该 channel，需此步
}