5. 并发模式与最佳实践

学习常见的并发模式和最佳实践，编写安全高效的并发代码。
本章将从简单到复杂，通过大量图解和实际案例，深入讲解 Go 并发编程的核心模式。

5.1 Worker Pool 模式

5.1.1 什么是 Worker Pool？

Worker Pool（工作池）是一种并发模式，通过预先创建一定数量的 worker goroutine 来处理任务，避免无限制地创建 goroutine 导致资源耗尽。

┌─────────────────────────────────────────────────────────────────┐
│                    Worker Pool 架构图                            │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   任务队列                                                      │
│  ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐ ┌─────┐                     │
│  │Task1│ │Task2│ │Task3│ │Task4│ │Task5│  ...               │
│  └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘ └──┬──┘                     │
│     │       │       │       │       │                          │
│     └───────┴───┬───┴───────┴───────┘                          │
│                 │                                               │
│                 ▼                                               │
│  ┌─────────────────────────────────────────────────────┐       │
│  │              Worker Pool                            │       │
│  │  ┌────────┐  ┌────────┐  ┌────────┐  ┌────────┐   │       │
│  │  │Worker 1│  │Worker 2│  │Worker 3│  │Worker 4│   │       │
│  │  └───┬────┘  └───┬────┘  └───┬────┘  └───┬────┘   │       │
│  └──────┼───────────┼───────────┼───────────┼─────────┘       │
│         │           │           │           │                  │
│         └───────────┴─────┬─────┴───────────┘                  │
│                           │                                     │
│                           ▼                                     │
│                      结果收集                                   │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

5.1.2 基础实现

让我们从最简单的 Worker Pool 开始：

package main

import (
    "fmt"
    "sync"
    "time"
)

// Job 任务定义
type Job struct {
    ID   int
    Data string
}

// Result 结果定义
type Result struct {
    JobID  int
    Output string
    Worker int
}

// worker 工作协程
func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for job := range jobs {
        // 模拟耗时操作
        time.Sleep(100 * time.Millisecond)
        
        // 处理任务
        output := fmt.Sprintf("Worker %d 处理了任务 %d: %s", id, job.ID, job.Data)
        
        // 发送结果
        results <- Result{
            JobID:  job.ID,
            Output: output,
            Worker: id,
        }
    }
}

func main() {
    const numWorkers = 3
    const numJobs = 10
    
    // 创建任务通道和结果通道
    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)
    
    var wg sync.WaitGroup
    
    // 启动 worker
    fmt.Println("启动 Worker Pool...")
    for w := 1; w <= numWorkers; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg)
    }
    
    // 发送任务
    fmt.Println("发送任务...")
    for j := 1; j <= numJobs; j++ {
        jobs <- Job{ID: j, Data: fmt.Sprintf("task-%d", j)}
    }
    close(jobs)  // 关闭任务通道，通知 worker 没有更多任务了
    
    // 等待所有 worker 完成并关闭结果通道
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // 收集结果
    fmt.Println("收集结果...")
    for result := range results {
        fmt.Printf("✓ %s\n", result.Output)
    }
    
    fmt.Println("所有任务完成!")
}

运行输出：

启动 Worker Pool...
发送任务...
收集结果...
✓ Worker 1 处理了任务 1: task-1
✓ Worker 2 处理了任务 2: task-2
✓ Worker 3 处理了任务 3: task-3
✓ Worker 1 处理了任务 4: task-4
✓ Worker 2 处理了任务 5: task-5
...
所有任务完成!

5.1.3 进阶：带限流的 Worker Pool

在实际应用中，我们通常需要控制任务的提交速率：

package main

import (
    "context"
    "fmt"
    "sync"
    "time"
)

// WorkerPool 工作池
type WorkerPool struct {
    workerCount int
    jobs        chan Job
    results     chan Result
    wg          sync.WaitGroup
    ctx         context.Context
    cancel      context.CancelFunc
}

// NewWorkerPool 创建工作池
func NewWorkerPool(workerCount, jobBufferSize int) *WorkerPool {
    ctx, cancel := context.WithCancel(context.Background())
    
    wp := &WorkerPool{
        workerCount: workerCount,
        jobs:        make(chan Job, jobBufferSize),
        results:     make(chan Result, jobBufferSize),
        ctx:         ctx,
        cancel:      cancel,
    }
    
    // 启动 worker
    wp.start()
    
    return wp
}

func (wp *WorkerPool) start() {
    for i := 1; i <= wp.workerCount; i++ {
        wp.wg.Add(1)
        go wp.worker(i)
    }
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    
    for {
        select {
        case <-wp.ctx.Done():
            fmt.Printf("Worker %d 收到退出信号\n", id)
            return
        case job, ok := <-wp.jobs:
            if !ok {
                fmt.Printf("Worker %d: 任务通道已关闭\n", id)
                return
            }
            
            // 处理任务
            result := wp.processJob(id, job)
            
            // 发送结果
            select {
            case wp.results <- result:
            case <-wp.ctx.Done():
                return
            }
        }
    }
}

func (wp *WorkerPool) processJob(workerID int, job Job) Result {
    // 模拟处理
    time.Sleep(100 * time.Millisecond)
    
    return Result{
        JobID:  job.ID,
        Output: fmt.Sprintf("处理完成：%s", job.Data),
        Worker: workerID,
    }
}

// Submit 提交任务
func (wp *WorkerPool) Submit(job Job) error {
    select {
    case wp.jobs <- job:
        return nil
    case <-wp.ctx.Done():
        return fmt.Errorf("工作池已关闭")
    default:
        return fmt.Errorf("任务队列已满")
    }
}

// SubmitWait 提交任务（阻塞等待）
func (wp *WorkerPool) SubmitWait(job Job) error {
    select {
    case wp.jobs <- job:
        return nil
    case <-wp.ctx.Done():
        return fmt.Errorf("工作池已关闭")
    }
}

// Results 获取结果通道
func (wp *WorkerPool) Results() <-chan Result {
    return wp.results
}

// Shutdown 关闭工作池
func (wp *WorkerPool) Shutdown() {
    wp.cancel()
    close(wp.jobs)
    wp.wg.Wait()
    close(wp.results)
}

func main() {
    // 创建工作池
    pool := NewWorkerPool(4, 100)
    
    // 启动结果处理
    go func() {
        for result := range pool.Results() {
            fmt.Printf("结果：%+v\n", result)
        }
    }()
    
    // 提交任务
    for i := 1; i <= 20; i++ {
        job := Job{ID: i, Data: fmt.Sprintf("task-%d", i)}
        if err := pool.SubmitWait(job); err != nil {
            fmt.Printf("提交失败：%v\n", err)
        }
    }
    
    // 等待一下让任务处理完成
    time.Sleep(time.Second)
    
    // 关闭工作池
    pool.Shutdown()
    fmt.Println("工作池已关闭")
}

5.1.4 实战：Kubernetes 中的 Worker Pool

Kubernetes 的 Controller 模式是 Worker Pool 的经典应用：

// 简化的 Kubernetes Controller 实现
type Controller struct {
    queue    workqueue.RateLimitingInterface
    workers  int
    informer cache.SharedIndexInformer
}

func (c *Controller) Run(workers int, stopCh <-chan struct{}) {
    defer utilruntime.HandleCrash()
    
    // 启动多个 worker
    for i := 0; i < workers; i++ {
        go wait.Until(c.runWorker, time.Second, stopCh)
    }
    
    <-stopCh
    c.queue.ShutDown()
}

func (c *Controller) runWorker() {
    for c.processNextWorkItem() {
    }
}

func (c *Controller) processNextWorkItem() bool {
    obj, shutdown := c.queue.Get()
    if shutdown {
        return false
    }
    
    // 使用 defer 确保任务完成标记
    defer c.queue.Done(obj)
    
    // 处理任务
    err := c.syncHandler(obj.(string))
    if err != nil {
        // 处理失败，重新加入队列
        c.queue.AddRateLimited(obj)
        return true
    }
    
    // 处理成功
    c.queue.Forget(obj)
    return true
}

5.1.5 最佳实践

// ✓ 好的实践：使用 context 控制生命周期
func NewWorkerPool(ctx context.Context, workerCount int) *WorkerPool {
    wp := &WorkerPool{
        ctx: ctx,
        // ...
    }
    wp.start()
    return wp
}

// ✓ 好的实践：优雅关闭
func (wp *WorkerPool) Shutdown() {
    wp.cancel()      // 1. 发送取消信号
    close(wp.jobs)   // 2. 关闭输入
    wp.wg.Wait()     // 3. 等待 worker 完成
    close(wp.results)// 4. 关闭输出
}

// ✗ 坏的实践：goroutine 泄漏
func BadWorkerPool() {
    jobs := make(chan Job)
    // 没有关闭通道，worker 永远阻塞
    go func() {
        for job := range jobs {
            process(job)
        }
    }()
    // 忘记 close(jobs)
}

5.2 生产者 - 消费者模式

5.2.1 模式说明

生产者 - 消费者模式通过 channel 解耦生产和消费过程：

┌─────────────────────────────────────────────────────────────────┐
│               生产者 - 消费者模式                                │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│   生产者 1      生产者 2      生产者 3                         │
│  ┌────────┐   ┌────────┐   ┌────────┐                        │
│  │ Produce│   │ Produce│   │ Produce│                        │
│  └───┬────┘   └───┬────┘   └───┬────┘                        │
│      │            │            │                               │
│      └────────────┼────────────┘                               │
│                   │                                             │
│                   ▼                                             │
│         ┌───────────────────┐                                  │
│         │   Channel (Buffer)│                                  │
│         │  [ ][ ][ ][ ][ ]  │                                  │
│         └─────────┬─────────┘                                  │
│                   │                                             │
│      ┌────────────┼────────────┐                               │
│      │            │            │                               │
│      ▼            ▼            ▼                               │
│  ┌────────┐   ┌────────┐   ┌────────┐                        │
│  │ Consume│   │ Consume│   │ Consume│                        │
│  └────────┘   └────────┘   └────────┘                        │
│   消费者 1      消费者 2      消费者 3                         │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

5.2.2 基础实现

package main

import (
    "fmt"
    "sync"
    "time"
)

func main() {
    ch := make(chan int, 10)
    var wg sync.WaitGroup
    
    // 启动多个消费者
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go consumer(i, ch, &wg)
    }
    
    // 启动生产者
    wg.Add(1)
    go producer(ch, &wg)
    
    wg.Wait()
    fmt.Println("所有任务完成!")
}

// producer 生产者
func producer(ch chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    defer close(ch)  // 生产完成后关闭通道
    
    for i := 1; i <= 10; i++ {
        ch <- i
        fmt.Printf("📦 生产：%d\n", i)
        time.Sleep(50 * time.Millisecond)
    }
}

// consumer 消费者
func consumer(id int, ch <-chan int, wg *sync.WaitGroup) {
    defer wg.Done()
    
    for item := range ch {
        // 模拟消费
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("🔄 消费者 %d 消费：%d\n", id, item)
    }
}

5.2.3 多级管道

复杂场景下，我们可以将多个生产者 - 消费者串联成管道：

package main

import (
    "fmt"
    "sync"
)

// 阶段 1: 生成数据
func generator(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// 阶段 2: 平方计算
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// 阶段 3: 过滤
func filter(in <-chan int, predicate func(int) bool) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if predicate(n) {
                out <- n
            }
        }
    }()
    return out
}

// 阶段 4: 汇总
func sum(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        total := 0
        for n := range in {
            total += n
        }
        out <- total
    }()
    return out
}

func main() {
    // 构建管道
    pipeline := generator(1, 2, 3, 4, 5)
    pipeline = square(pipeline)
    pipeline = filter(pipeline, func(n int) bool { return n > 10 })
    pipeline = sum(pipeline)
    
    // 获取结果
    result := <-pipeline
    fmt.Printf("最终结果：%d\n", result)  // 4 + 9 + 16 + 25 = 54
}

5.2.4 实战：日志处理系统

package main

import (
    "fmt"
    "sync"
    "time"
)

// LogEntry 日志条目
type LogEntry struct {
    Timestamp time.Time
    Level     string
    Message   string
}

// LogProcessor 日志处理器
type LogProcessor struct {
    input  chan LogEntry
    output chan string
    wg     sync.WaitGroup
}

func NewLogProcessor(workers int) *LogProcessor {
    lp := &LogProcessor{
        input:  make(chan LogEntry, 1000),
        output: make(chan string, 1000),
    }
    
    // 启动多个处理 worker
    for i := 0; i < workers; i++ {
        lp.wg.Add(1)
        go lp.processWorker(i)
    }
    
    // 启动输出 worker
    lp.wg.Add(1)
    go lp.outputWorker()
    
    return lp
}

func (lp *LogProcessor) processWorker(id int) {
    defer lp.wg.Done()
    
    for entry := range lp.input {
        // 处理日志
        formatted := fmt.Sprintf("[%s] %s: %s (worker-%d)",
            entry.Timestamp.Format("15:04:05"),
            entry.Level,
            entry.Message,
            id)
        
        lp.output <- formatted
    }
}

func (lp *LogProcessor) outputWorker() {
    defer lp.wg.Done()
    
    for formatted := range lp.output {
        // 输出到文件/控制台
        fmt.Println(formatted)
    }
}

func (lp *LogProcessor) Submit(entry LogEntry) {
    lp.input <- entry
}

func (lp *LogProcessor) Shutdown() {
    close(lp.input)
    lp.wg.Wait()
    close(lp.output)
}

func main() {
    processor := NewLogProcessor(3)
    
    // 模拟日志输入
    for i := 0; i < 10; i++ {
        processor.Submit(LogEntry{
            Timestamp: time.Now(),
            Level:     "INFO",
            Message:   fmt.Sprintf("Log message %d", i),
        })
    }
    
    time.Sleep(time.Second)
    processor.Shutdown()
}

5.3 发布 - 订阅模式

5.3.1 模式说明

发布 - 订阅（Pub/Sub）模式允许消息的发送者（发布者）和接收者（订阅者）解耦：

┌─────────────────────────────────────────────────────────────────┐
│                    发布 - 订阅模式                               │
├─────────────────────────────────────────────────────────────────┤
│                                                                 │
│                    ┌─────────────┐                             │
│     发布者 1 ─────►│             │                             │
│                    │   Broker    │                             │
│     发布者 2 ─────►│   (Topic)   │                             │
│                    │             │                             │
│     发布者 3 ─────►│             │                             │
│                    └──────┬──────┘                             │
│                           │                                     │
│         ┌─────────────────┼─────────────────┐                  │
│         │                 │                 │                  │
│         ▼                 ▼                 ▼                  │
│    ┌────────┐       ┌────────┐       ┌────────┐              │
│    │订阅者 1 │       │订阅者 2 │       │订阅者 3 │              │
│    │ Topic A│       │Topic A,B│      │Topic B │              │
│    └────────┘       └────────┘       └────────┘              │
│                                                                 │
└─────────────────────────────────────────────────────────────────┘

5.3.2 基础实现

package main

import (
    "fmt"
    "sync"
    "time"
)

// PubSub 发布订阅系统
type PubSub struct {
    mu    sync.RWMutex
    subs  map[string][]chan string
}

// NewPubSub 创建发布订阅系统
func NewPubSub() *PubSub {
    return &PubSub{
        subs: make(map[string][]chan string),
    }
}

// Subscribe 订阅主题
func (ps *PubSub) Subscribe(topic string) <-chan string {
    ps.mu.Lock()
    defer ps.mu.Unlock()
    
    ch := make(chan string, 100)
    ps.subs[topic] = append(ps.subs[topic], ch)
    
    return ch
}

// Unsubscribe 取消订阅
func (ps *PubSub) Unsubscribe(topic string, ch <-chan string) {
    ps.mu.Lock()
    defer ps.mu.Unlock()
    
    subs := ps.subs[topic]
    for i, sub := range subs {
        if sub == ch {
            ps.subs[topic] = append(subs[:i], subs[i+1:]...)
            close(ch)
            return
        }
    }
}

// Publish 发布消息
func (ps *PubSub) Publish(topic string, msg string) {
    ps.mu.RLock()
    defer ps.mu.RUnlock()
    
    for _, ch := range ps.subs[topic] {
        select {
        case ch <- msg:
        default:
            // 通道已满，跳过
            fmt.Printf("警告：主题 %s 的订阅者通道已满\n", topic)
        }
    }
}

// Broadcast 广播消息到所有主题
func (ps *PubSub) Broadcast(msg string) {
    ps.mu.RLock()
    defer ps.mu.RUnlock()
    
    for topic := range ps.subs {
        ps.Publish(topic, msg)
    }
}

func main() {
    ps := NewPubSub()
    
    // 订阅者 1: 订阅 news
    sub1 := ps.Subscribe("news")
    go func() {
        for msg := range sub1 {
            fmt.Printf("📰 订阅者 1 收到新闻：%s\n", msg)
        }
    }()
    
    // 订阅者 2: 订阅 news 和 sports
    sub2 := ps.Subscribe("news")
    sub2Sports := ps.Subscribe("sports")
    go func() {
        for msg := range sub2 {
            fmt.Printf("📰 订阅者 2 收到新闻：%s\n", msg)
        }
    }()
    go func() {
        for msg := range sub2Sports {
            fmt.Printf("⚽ 订阅者 2 收到体育：%s\n", msg)
        }
    }()
    
    // 订阅者 3: 订阅 sports
    sub3 := ps.Subscribe("sports")
    go func() {
        for msg := range sub3 {
            fmt.Printf("⚽ 订阅者 3 收到体育：%s\n", msg)
        }
    }()
    
    // 发布消息
    ps.Publish("news", "重大新闻：Go 2.0 发布!")
    ps.Publish("sports", "体育快讯：中国队夺冠!")
    ps.Publish("news", "突发新闻：...")
    
    time.Sleep(time.Second)
}

5.3.3 进阶：带缓冲的 PubSub

// 带缓冲的发布订阅
type BufferedPubSub struct {
    mu           sync.RWMutex
    subs         map[string][]*subscriber
    defaultBuf   int
}

type subscriber struct {
    ch   chan string
    done chan struct{}
}

func NewBufferedPubSub(bufSize int) *BufferedPubSub {
    return &BufferedPubSub{
        subs:       make(map[string][]*subscriber),
        defaultBuf: bufSize,
    }
}

func (ps *BufferedPubSub) Subscribe(topic string) <-chan string {
    ps.mu.Lock()
    defer ps.mu.Unlock()
    
    sub := &subscriber{
        ch:   make(chan string, ps.defaultBuf),
        done: make(chan struct{}),
    }
    
    ps.subs[topic] = append(ps.subs[topic], sub)
    return sub.ch
}

func (ps *BufferedPubSub) Publish(topic string, msg string) int {
    ps.mu.RLock()
    defer ps.mu.RUnlock()
    
    sent := 0
    for _, sub := range ps.subs[topic] {
        select {
        case sub.ch <- msg:
            sent++
        case <-sub.done:
            // 订阅者已取消
        default:
            // 通道已满
        }
    }
    return sent
}

5.3.4 实战：事件总线

// EventBus 事件总线
type EventBus struct {
    *PubSub
}

type Event struct {
    Type      string
    Payload   interface{}
    Timestamp time.Time
}

func NewEventBus() *EventBus {
    return &EventBus{
        PubSub: NewPubSub(),
    }
}

func (eb *EventBus) Subscribe(eventType string, handler func(Event)) func() {
    ch := eb.PubSub.Subscribe(eventType)
    done := make(chan struct{})
    
    go func() {
        for {
            select {
            case msg := <-ch:
                // 这里需要反序列化
                var event Event
                // ... 解析 msg
                handler(event)
            case <-done:
                return
            }
        }
    }()
    
    return func() {
        close(done)
        eb.PubSub.Unsubscribe(eventType, ch)
    }
}

func (eb *EventBus) Publish(eventType string, payload interface{}) {
    event := Event{
        Type:      eventType,
        Payload:   payload,
        Timestamp: time.Now(),
    }
    // 序列化 event
    msg := fmt.Sprintf("%+v", event)
    eb.PubSub.Publish(eventType, msg)
}

5.4 并发安全模式

5.4.1 使用 Mutex 保护共享状态

// SafeMap 线程安全的 Map
type SafeMap struct {
    mu   sync.RWMutex
    data map[string]string
}

func NewSafeMap() *SafeMap {
    return &SafeMap{
        data: make(map[string]string),
    }
}

// Get 获取值（读锁）
func (sm *SafeMap) Get(key string) (string, bool) {
    sm.mu.RLock()
    defer sm.mu.RUnlock()
    val, ok := sm.data[key]
    return val, ok
}

// Set 设置值（写锁）
func (sm *SafeMap) Set(key, value string) {
    sm.mu.Lock()
    defer sm.mu.Unlock()
    sm.data[key] = value
}

// Delete 删除值
func (sm *SafeMap) Delete(key string) {
    sm.mu.Lock()
    defer sm.mu.Unlock()
    delete(sm.data, key)
}

// Len 获取长度
func (sm *SafeMap) Len() int {
    sm.mu.RLock()
    defer sm.mu.RUnlock()
    return len(sm.data)
}

// Range 遍历
func (sm *SafeMap) Range(fn func(key, value string) bool) {
    sm.mu.RLock()
    defer sm.mu.RUnlock()
    
    for k, v := range sm.data {
        if !fn(k, v) {
            return
        }
    }
}

5.4.2 使用 Channel 序列化访问

// SafeCounter 使用 channel 实现线程安全计数器
type SafeCounter struct {
    requests chan func()
    count    int
}

func NewSafeCounter() *SafeCounter {
    c := &SafeCounter{
        requests: make(chan func(), 100),
    }
    go c.process()
    return c
}

func (c *SafeCounter) process() {
    for req := range c.requests {
        req()
    }
}

func (c *SafeCounter) Increment() {
    c.requests <- func() {
        c.count++
    }
}

func (c *SafeCounter) Decrement() {
    c.requests <- func() {
        c.count--
    }
}

func (c *SafeCounter) Value() int {
    result := make(chan int)
    c.requests <- func() {
        result <- c.count
    }
    return <-result
}

func (c *SafeCounter) Close() {
    close(c.requests)
}

5.4.3 使用 sync.Map

// sync.Map 适用于读多写少的场景
var cache sync.Map

// 存储
cache.Store("key", "value")

// 读取
val, ok := cache.Load("key")

// 读取或存储
val, loaded := cache.LoadOrStore("key", "value")

// 删除
cache.Delete("key")

// 遍历
cache.Range(func(key, value interface{}) bool {
    fmt.Printf("%v: %v\n", key, value)
    return true
})

5.4.4 模式对比

模式	适用场景	优点	缺点
Mutex	通用场景	简单直观	可能有锁竞争
RWMutex	读多写少	读操作并发	写操作阻塞所有读
Channel	需要通信	天然同步	性能略低
sync.Map	读多写少	无锁读取	写操作慢

5.5 错误处理

5.5.1 使用 errgroup

package main

import (
    "context"
    "errors"
    "fmt"
    "golang.org/x/sync/errgroup"
    "time"
)

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    
    // 任务 1
    g.Go(func() error {
        return doTask1(ctx)
    })
    
    // 任务 2
    g.Go(func() error {
        return doTask2(ctx)
    })
    
    // 任务 3
    g.Go(func() error {
        return doTask3(ctx)
    })
    
    // 等待所有任务完成
    if err := g.Wait(); err != nil {
        fmt.Printf("有任务失败：%v\n", err)
    } else {
        fmt.Println("所有任务成功!")
    }
}

func doTask1(ctx context.Context) error {
    time.Sleep(100 * time.Millisecond)
    return nil
}

func doTask2(ctx context.Context) error {
    time.Sleep(200 * time.Millisecond)
    return errors.New("任务 2 失败")
}

func doTask3(ctx context.Context) error {
    <-ctx.Done()  // 任务 2 失败后，ctx 被取消
    return ctx.Err()
}

输出：

有任务失败：任务 2 失败

5.5.2 带限流的 errgroup

package main

import (
    "context"
    "fmt"
    "golang.org/x/sync/errgroup"
    "time"
)

func main() {
    g, ctx := errgroup.WithContext(context.Background())
    
    // 限制并发数为 3
    semaphore := make(chan struct{}, 3)
    
    for i := 0; i < 10; i++ {
        i := i
        g.Go(func() error {
            // 获取信号量
            select {
            case semaphore <- struct{}{}:
            case <-ctx.Done():
                return ctx.Err()
            }
            defer func() { <-semaphore }()
            
            return doWork(ctx, i)
        })
    }
    
    if err := g.Wait(); err != nil {
        fmt.Printf("错误：%v\n", err)
    }
}

func doWork(ctx context.Context, i int) error {
    fmt.Printf("任务 %d 开始\n", i)
    time.Sleep(100 * time.Millisecond)
    fmt.Printf("任务 %d 完成\n", i)
    return nil
}

5.5.3 错误收集器

// ErrorCollector 错误收集器
type ErrorCollector struct {
    mu   sync.Mutex
    errs []error
}

func (ec *ErrorCollector) Add(err error) {
    if err != nil {
        ec.mu.Lock()
        defer ec.mu.Unlock()
        ec.errs = append(ec.errs, err)
    }
}

func (ec *ErrorCollector) Errors() []error {
    ec.mu.Lock()
    defer ec.mu.Unlock()
    return append([]error(nil), ec.errs...)
}

func (ec *ErrorCollector) Error() error {
    ec.mu.Lock()
    defer ec.mu.Unlock()
    
    if len(ec.errs) == 0 {
        return nil
    }
    
    // 返回所有错误
    var msg string
    for i, err := range ec.errs {
        if i > 0 {
            msg += "; "
        }
        msg += err.Error()
    }
    return errors.New(msg)
}

// 使用示例
func main() {
    var ec ErrorCollector
    var wg sync.WaitGroup
    
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            err := doWork(id)
            ec.Add(err)
        }(i)
    }
    
    wg.Wait()
    
    if err := ec.Error(); err != nil {
        fmt.Printf("有错误发生：%v\n", err)
    }
}

5.6 性能优化

5.6.1 使用 sync.Pool 减少分配

package main

import (
    "bytes"
    "fmt"
    "sync"
)

// 创建对象池
var bufferPool = sync.Pool{
    New: func() interface{} {
        return new(bytes.Buffer)
    },
}

// 从池中获取对象
func getBuffer() *bytes.Buffer {
    return bufferPool.Get().(*bytes.Buffer)
}

// 归还对象到池
func putBuffer(buf *bytes.Buffer) {
    buf.Reset()
    bufferPool.Put(buf)
}

func main() {
    // 使用示例
    buf := getBuffer()
    buf.WriteString("Hello, World!")
    fmt.Println(buf.String())
    putBuffer(buf)
    
    // 再次获取，复用内存
    buf2 := getBuffer()
    buf2.WriteString("Reused buffer!")
    fmt.Println(buf2.String())
    putBuffer(buf2)
}

5.6.2 避免 Goroutine 泄漏

// ✓ 正确：使用 context 控制生命周期
func worker(ctx context.Context, input <-chan int) {
    for {
        select {
        case <-ctx.Done():
            fmt.Println("worker 退出")
            return
        case val, ok := <-input:
            if !ok {
                return
            }
            process(val)
        }
    }
}

// ✓ 正确：使用超时
func workerWithTimeout(input <-chan int, timeout time.Duration) {
    for {
        select {
        case val, ok := <-input:
            if !ok {
                return
            }
            process(val)
        case <-time.After(timeout):
            fmt.Println("超时退出")
            return
        }
    }
}

// ✗ 错误：没有退出机制
func badWorker(input <-chan int) {
    for val := range input {
        process(val)
        // 如果 input 永远不关闭，这个 goroutine 永远不退出
    }
}

5.6.3 减少锁竞争

// ✗ 锁粒度过大
type BadCache struct {
    mu   sync.Mutex
    data map[string]string
}

func (c *BadCache) Get(key string) string {
    c.mu.Lock()  // 读也加锁
    defer c.mu.Unlock()
    return c.data[key]
}

// ✓ 使用 RWMutex
type GoodCache struct {
    mu   sync.RWMutex
    data map[string]string
}

func (c *GoodCache) Get(key string) string {
    c.mu.RLock()  // 读锁，可并发
    defer c.mu.RUnlock()
    return c.data[key]
}

// ✓ 分段锁（Sharding）
type ShardedMap struct {
    shards [16]*SafeMap
}

func (sm *ShardedMap) Get(key string) (string, bool) {
    shard := sm.getShard(key)
    return shard.Get(key)
}

func (sm *ShardedMap) getShard(key string) *SafeMap {
    hash := hashKey(key)
    return sm.shards[hash%16]
}

5.7 实战案例

5.7.1 并发文件处理

package main

import (
    "fmt"
    "io/ioutil"
    "os"
    "path/filepath"
    "sync"
)

// FileProcessor 文件处理器
type FileProcessor struct {
    workers int
    wg      sync.WaitGroup
}

func NewFileProcessor(workers int) *FileProcessor {
    return &FileProcessor{workers: workers}
}

func (fp *FileProcessor) ProcessDir(dir string) error {
    // 收集所有文件
    var files []string
    err := filepath.Walk(dir, func(path string, info os.FileInfo, err error) error {
        if err != nil {
            return err
        }
        if !info.IsDir() {
            files = append(files, path)
        }
        return nil
    })
    if err != nil {
        return err
    }
    
    // 创建任务通道
    jobs := make(chan string, len(files))
    results := make(chan Result, len(files))
    
    // 启动 worker
    for i := 0; i < fp.workers; i++ {
        fp.wg.Add(1)
        go fp.worker(jobs, results)
    }
    
    // 发送任务
    for _, file := range files {
        jobs <- file
    }
    close(jobs)
    
    // 等待完成
    go func() {
        fp.wg.Wait()
        close(results)
    }()
    
    // 收集结果
    for result := range results {
        if result.Err != nil {
            fmt.Printf("处理 %s 失败：%v\n", result.File, result.Err)
        } else {
            fmt.Printf("处理 %s 成功，大小：%d\n", result.File, result.Size)
        }
    }
    
    return nil
}

func (fp *FileProcessor) worker(jobs <-chan string, results chan<- Result) {
    defer fp.wg.Done()
    
    for file := range jobs {
        info, err := os.Stat(file)
        if err != nil {
            results <- Result{File: file, Err: err}
            continue
        }
        
        results <- Result{
            File: file,
            Size: info.Size(),
        }
    }
}

type Result struct {
    File string
    Size int64
    Err  error
}

func main() {
    processor := NewFileProcessor(4)
    processor.ProcessDir(".")
}

5.7.2 并发 HTTP 请求

package main

import (
    "context"
    "fmt"
    "io/ioutil"
    "net/http"
    "sync"
    "time"
)

// HTTPClient 并发 HTTP 客户端
type HTTPClient struct {
    client  *http.Client
    timeout time.Duration
}

func NewHTTPClient(timeout time.Duration) *HTTPClient {
    return &HTTPClient{
        client: &http.Client{Timeout: timeout},
        timeout: timeout,
    }
}

// FetchAll 并发获取多个 URL
func (hc *HTTPClient) FetchAll(urls []string, concurrency int) ([]Result, error) {
    jobs := make(chan string, len(urls))
    results := make(chan Result, len(urls))
    
    var wg sync.WaitGroup
    
    // 启动 worker
    for i := 0; i < concurrency; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for url := range jobs {
                results <- hc.fetch(url)
            }
        }()
    }
    
    // 发送任务
    for _, url := range urls {
        jobs <- url
    }
    close(jobs)
    
    // 等待完成
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // 收集结果
    var allResults []Result
    for result := range results {
        allResults = append(allResults, result)
    }
    
    return allResults, nil
}

func (hc *HTTPClient) fetch(url string) Result {
    ctx, cancel := context.WithTimeout(context.Background(), hc.timeout)
    defer cancel()
    
    req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
    resp, err := hc.client.Do(req)
    
    result := Result{URL: url}
    if err != nil {
        result.Err = err
        return result
    }
    defer resp.Body.Close()
    
    body, _ := ioutil.ReadAll(resp.Body)
    result.StatusCode = resp.StatusCode
    result.Body = string(body)
    
    return result
}

type Result struct {
    URL        string
    StatusCode int
    Body       string
    Err        error
}

func main() {
    client := NewHTTPClient(5 * time.Second)
    
    urls := []string{
        "https://api.github.com",
        "https://api.example.com",
        "https://httpbin.org/get",
    }
    
    results, _ := client.FetchAll(urls, 3)
    for _, r := range results {
        if r.Err != nil {
            fmt.Printf("%s: 错误 %v\n", r.URL, r.Err)
        } else {
            fmt.Printf("%s: 状态 %d\n", r.URL, r.StatusCode)
        }
    }
}

5.8 本章小结

┌─────────────────────────────────────────────────────────────┐
│                      本章总结                                │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ✓ Worker Pool: 控制并发数量，避免资源耗尽                │
│     └── 基础实现 → 带限流 → Kubernetes 实践              │
│                                                             │
│  ✓ 生产者 - 消费者：解耦生产和消费                       │
│     └── 基础模式 → 多级管道 → 日志处理系统              │
│                                                             │
│  ✓ 发布 - 订阅：一对多通信                                │
│     └── 基础 PubSub → 带缓冲 → 事件总线                │
│                                                             │
│  ✓ 并发安全：保护共享状态                                │
│     └── Mutex → Channel → sync.Map → 对比选择          │
│                                                             │
│  ✓ 错误处理：优雅的错误管理                             │
│     └── errgroup → 限流 → 错误收集器                  │
│                                                             │
│  ✓ 性能优化：提升并发效率                               │
│     └── sync.Pool → 避免泄漏 → 减少锁竞争             │
│                                                             │
│  ✓ 实战案例：文件处理、HTTP 请求                        │
│                                                             │
└─────────────────────────────────────────────────────────────┘

关键要点

选择合适的模式：根据场景选择 Worker Pool、Pub/Sub 等模式
注意资源管理：及时关闭 channel，避免 goroutine 泄漏
优雅的错误处理：使用 errgroup 等工具管理并发错误
性能优化：使用 sync.Pool、减少锁竞争等手段提升性能
使用 context：始终使用 context 控制 goroutine 生命周期

上一章：← Context 深度解析 | 下一章：内存模型与竞态检测 →

5. 并发模式与最佳实践 ​

目录 ​

5.1 Worker Pool 模式 ​

5.1.1 什么是 Worker Pool？ ​

5.1.2 基础实现 ​

5.1.3 进阶：带限流的 Worker Pool ​

5.1.4 实战：Kubernetes 中的 Worker Pool ​

5.1.5 最佳实践 ​

5.2 生产者 - 消费者模式 ​

5.2.1 模式说明 ​

5.2.2 基础实现 ​

5.2.3 多级管道 ​

5.2.4 实战：日志处理系统 ​

5.3 发布 - 订阅模式 ​

5.3.1 模式说明 ​

5.3.2 基础实现 ​

5.3.3 进阶：带缓冲的 PubSub ​

5.3.4 实战：事件总线 ​

5.4 并发安全模式 ​

5.4.1 使用 Mutex 保护共享状态 ​

5.4.2 使用 Channel 序列化访问 ​

5.4.3 使用 sync.Map ​

5.4.4 模式对比 ​

5.5 错误处理 ​

5.5.1 使用 errgroup ​

5.5.2 带限流的 errgroup ​

5.5.3 错误收集器 ​

5.6 性能优化 ​

5.6.1 使用 sync.Pool 减少分配 ​

5.6.2 避免 Goroutine 泄漏 ​

5.6.3 减少锁竞争 ​

5.7 实战案例 ​

5.7.1 并发文件处理 ​

5.7.2 并发 HTTP 请求 ​

5.8 本章小结 ​

关键要点 ​