Go 中多生产者-多消费者模型下的竞态与同步实践指南

心靈之曲

发布时间：2026-02-05 08:43:02

223人浏览过

来源于php中文网

原创

Go 中多生产者-多消费者模型下的竞态与同步实践指南

本文详解 go 语言中多生产者/多消费者并发模型的典型陷阱——全局变量非原子操作引发的数据竞争，演示如何用 `atomic` 替代裸读写，并通过通道协调实现线程安全的序列号分发。

在 Go 并发编程中，「多个 goroutine 同时读写同一变量」是导致不可预测行为的根源之一。你提供的代码看似能稳定输出 1 到 1000 的递增序列，实则掩盖了一个严重问题：seq++（即 seq = seq + 1）不是原子操作，它包含「读取 → 计算 → 写入」三步，在无同步保护下极易发生数据竞争（data race）。

? 为什么没看到重复数字？——表象背后的偶然性

你的程序中：

requestChan 是无缓冲通道，所有消费者发送请求时会阻塞，直到某个 generateStuff goroutine 从该通道接收；
generatorChan 同样无缓冲，生产者生成结果后必须等待消费者接收才继续；
这种双重通道同步形成了隐式串行化：即使有 20 个生产者，实际任一时刻往往仅有一个 goroutine 在执行 seq++ —— 尤其在 GOMAXPROCS=1（默认单 OS 线程）环境下，调度器倾向于让一个 goroutine 完成整个循环，从而“侥幸”避免了竞态暴露。

但这只是运气好，而非正确。一旦启用多线程（如 GOMAXPROCS=4）、增加负载或稍作扰动（如插入 runtime.Gosched()），竞态便会显现：seq 可能丢失自增、重复赋值，甚至导致程序崩溃。

✅ 正确做法：用原子操作替代共享变量修改

Go 标准库 sync/atomic 提供了对基础类型的安全原子操作。将 seq++ 替换为：

苏打办公

360旗下的办公工具导航，优质海量工具

下载

s := atomic.AddUint64(&seq, 1) // 原子递增并返回新值

这确保无论多少 goroutine 并发调用，seq 的更新都严格按顺序执行，绝无丢失或覆盖。

? 完整可运行示例（含日志与资源清理）

以下为优化后的生产者-消费者模型，已消除竞态、增强可观测性，并显式关闭通道防止 goroutine 泄漏：

package main

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

var seq uint64 = 0
var generatorChan = make(chan uint64, 10) // 建议加小缓冲提升吞吐
var 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 {
        s := atomic.AddUint64(&seq, 1) // ✅ 线程安全递增
        log.Printf("[Gen %2d] Req %3d → Seq %d", genID, reqID, s)
        generatorChan <- s
    }
}

func worker(id int, work *sync.WaitGroup) {
    defer work.Done()
    for i := 0; i < 5; i++ {
        requestChan <- uint64(id)
        result := <-generatorChan
        log.Printf("\t[Wrk %3d] Got %4d", id, result)
        time.Sleep(1 * time.Millisecond) // 模拟处理延迟，放大并发可见性
    }
}

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

    const (
        numGen    = 20
        numWorker = 200
    )

    var wg sync.WaitGroup

    // 启动生产者
    for i := 0; i < numGen; i++ {
        go generator(i)
    }

    // 启动消费者
    wg.Add(numWorker)
    for i := 0; i < numWorker; i++ {
        go worker(i, &wg)
    }

    // 等待所有消费者完成
    wg.Wait()

    // 安全关闭通道（仅关闭发送端）
    close(requestChan)

    // 可选：等待生产者自然退出（因 range 会自动检测 closed）
    time.Sleep(10 * time.Millisecond)
}