搬砖程序员带你飞
搬砖程序员带你飞

Go 语言并发模式详解

lpflpf 收录于 Golang
 约 2800 字 预计阅读 6 分钟 

Go 语言以其强大的并发特性而闻名。本文将详细介绍 Go 语言中常见的并发模式，帮助你更好地理解和使用 Go 的并发特性。

1. 基本的 Goroutine 和 Channel 模式

生产者-消费者模式是并发编程中最基础和最常用的模式之一。这种模式通过将数据生产和消费解耦，可以更好地控制数据流和处理速度。

1.1 生产者-消费者模式

工作流程：

graph LR
    P1[生产者1] --> C[Channel]
    P2[生产者2] --> C
    C --> CO1[消费者1]
    C --> CO2[消费者2]
    style C fill:#f9f,stroke:#333,stroke-width:2px

特点：

  • 解耦数据生产和消费逻辑
  • 通过 channel 实现数据的安全传递
  • 可以控制数据处理的速度和顺序
  • 支持多生产者和多消费者场景

适用场景：

  • 数据流处理，如日志收集和处理
  • 任务队列系统
  • 数据管道处理
  • 异步处理系统

示例代码：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

func producer(ch chan<- int) {
    for i := 0; i < 10; i++ {
        ch <- i
    }
    close(ch)
}

func consumer(ch <-chan int) {
    for num := range ch {
        fmt.Println("消费:", num)
    }
}

func main() {
    ch := make(chan int)
    go producer(ch)
    consumer(ch)
}

1.2 扇入模式（Fan-in）

工作流程：

graph LR
    A[输入源1] --> D[合并通道]
    B[输入源2] --> D
    C[输入源3] --> D
    D --> E[输出]
    style D fill:#f9f,stroke:#333,stroke-width:2px

特点：

  • 将多个数据源合并到一个输出通道
  • 使用 WaitGroup 确保所有输入源处理完成
  • 支持动态数量的输入源
  • 保持数据顺序的一致性

适用场景：

  • 合并多个数据源的处理结果
  • 聚合多个服务的响应
  • 并行计算结果的汇总
  • 多渠道数据采集系统

将多个输入源合并到一个输出源：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

func merge(cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup
    
    for _, c := range cs {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for n := range c {
                out <- n
            }
        }(c)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

1.3 扇出模式（Fan-out）

工作流程：

graph LR
    A[输入] --> D[分发通道]
    D --> B[处理器1]
    D --> C[处理器2]
    D --> E[处理器3]
    style D fill:#f9f,stroke:#333,stroke-width:2px

特点：

  • 将工作负载分散到多个处理单元
  • 实现并行处理提高效率
  • 可动态调整处理单元数量
  • 适合 CPU 密集型任务

适用场景：

  • 并行数据处理
  • 批量任务处理
  • 负载均衡
  • 大规模计算任务的分解

示例实现：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

func fanOut(in <-chan int, n int) []<-chan int {
    outputs := make([]<-chan int, n)
    for i := 0; i < n; i++ {
        ch := make(chan int)
        outputs[i] = ch
        go func(ch chan<- int) {
            defer close(ch)
            for num := range in {
                ch <- num * 2 // 示例处理
            }
        }(ch)
    }
    return outputs
}

2. 超时和取消模式

超时和取消是并发程序中的重要控制机制，可以避免资源浪费和程序阻塞。Go 的 context 包提供了优雅的实现方式。

2.1 使用 Context 进行超时控制

工作流程：

sequenceDiagram
    participant M as Main
    participant W as Worker
    participant T as Timer
    M->>W: 启动任务
    M->>T: 设置超时
    alt 任务完成
        W->>M: 返回结果
    else 超时发生
        T->>M: 超时信号
        M->>W: 取消任务
    end

特点：

  • 支持超时控制和取消操作
  • 可以传递截止时间、取消信号和请求作用域的值
  • 支持父子 context 的层级关系
  • 优雅的错误处理机制

适用场景：

  • HTTP 请求超时控制
  • RPC 调用超时处理
  • 数据库查询超时限制
  • 分布式系统的请求追踪
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

func worker(ctx context.Context) error {
    done := make(chan struct{})
    
    go func() {
        // 模拟耗时操作
        time.Sleep(2 * time.Second)
        done <- struct{}{}
    }()
    
    select {
    case <-done:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), time.Second)
    defer cancel()
    
    if err := worker(ctx); err != nil {
        fmt.Println("操作超时:", err)
    }
}

2.2 优雅退出模式

特点：

  • 确保资源正确释放
  • 等待进行中的任务完成
  • 避免数据丢失和状态不一致
  • 支持配置退出超时时间

适用场景：

  • 服务器程序的关闭处理
  • 后台任务的终止控制
  • 资源清理和持久化
  • 分布式系统的节点下线
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

func gracefulShutdown(stop <-chan struct{}) {
    ticker := time.NewTicker(time.Second)
    defer ticker.Stop()
    
    for {
        select {
        case <-ticker.C:
            fmt.Println("执行定时任务...")
        case <-stop:
            fmt.Println("正在清理资源...")
            return
        }
    }
}

3. 错误处理模式

并发程序中的错误处理需要特别注意，因为错误可能发生在任何 goroutine 中。合理的错误处理模式可以提高程序的可靠性。

3.1 错误传播

特点：

  • 通过 channel 传递错误信息
  • 支持异步操作的错误处理
  • 可以携带额外的错误上下文
  • 支持错误的聚合和过滤

适用场景：

  • 异步操作的错误处理
  • 分布式系统的错误收集
  • 并行任务的错误监控
  • 错误日志收集系统
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18

type Result struct {
    Value int
    Err   error
}

func asyncOperation() <-chan Result {
    ch := make(chan Result)
    go func() {
        defer close(ch)
        // 模拟可能出错的操作
        if rand.Float32() < 0.5 {
            ch <- Result{Err: fmt.Errorf("操作失败")}
            return
        }
        ch <- Result{Value: 42}
    }()
    return ch
}

3.2 错误组处理

特点：

  • 并行任务的错误同步处理
  • 支持多个 goroutine 的错误收集
  • 可以设置错误处理策略
  • 提供简洁的 API

适用场景：

  • 批量操作的错误处理
  • 并行任务的错误同步
  • 分布式事务的错误处理
  • 微服务调用的错误管理
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func parallelTasks() error {
    var errGroup sync.ErrorGroup
    
    // 添加多个任务
    errGroup.Go(func() error {
        // 任务1
        return nil
    })
    
    errGroup.Go(func() error {
        // 任务2
        return fmt.Errorf("任务2失败")
    })
    
    // 等待所有任务完成，返回第一个错误
    return errGroup.Wait()
}

4. 速率限制模式

速率限制是保护系统和资源的重要机制，Go 提供了多种实现速率限制的方式。

4.1 简单的速率限制

特点：

  • 固定时间间隔的处理控制
  • 简单易实现
  • 适合均匀负载场景
  • 低内存占用

适用场景：

  • API 访问频率限制
  • 资源下载速率控制
  • 消息推送频率限制
  • 定时任务控制
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

func rateLimiter() {
    ticker := time.NewTicker(200 * time.Millisecond)
    defer ticker.Stop()
    
    for i := 0; i < 10; i++ {
        <-ticker.C // 每200ms执行一次
        go func(i int) {
            fmt.Printf("处理请求 %d\n", i)
        }(i)
    }
}

4.2 令牌桶限流

特点：

  • 支持突发流量处理
  • 可配置的速率和容量
  • 平滑的限流效果
  • 支持动态调整参数

适用场景：

  • 高并发 API 限流
  • 网络带宽控制
  • 服务保护
  • 资源访问控制
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

func tokenBucket(rate int, capacity int) chan struct{} {
    tokens := make(chan struct{}, capacity)
    go func() {
        ticker := time.NewTicker(time.Second / time.Duration(rate))
        defer ticker.Stop()
        
        for range ticker.C {
            select {
            case tokens <- struct{}{}:
            default:
            }
        }
    }()
    return tokens
}

5. 池化模式

池化模式通过复用资源来提高程序性能和资源利用率。

5.1 Worker Pool

工作流程：

graph LR
    J[任务队列] --> P[Worker Pool]
    P --> W1[Worker 1]
    P --> W2[Worker 2]
    P --> W3[Worker 3]
    W1 --> R[结果收集]
    W2 --> R
    W3 --> R
    style P fill:#f9f,stroke:#333,stroke-width:2px
    style R fill:#9ff,stroke:#333,stroke-width:2px

特点：

  • 控制并发 goroutine 数量
  • 复用 goroutine 降低开销
  • 支持任务队列管理
  • 可动态调整池大小

适用场景：

  • 并发任务处理系统
  • 数据库连接池
  • HTTP 请求处理
  • 大规模并发计算
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14

func workerPool(numWorkers int, jobs <-chan int, results chan<- int) {
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for job := range jobs {
                results <- job * 2 // 示例工作处理
            }
        }()
    }
    wg.Wait()
    close(results)
}

总结

Go 语言的并发模式丰富多样，上述模式涵盖了日常开发中最常见的场景：

  1. 基本的生产者-消费者模式用于数据流处理
  2. 扇入扇出模式用于并行处理和数据聚合
  3. 超时和取消模式用于控制程序行为
  4. 错误处理模式确保程序的健壮性
  5. 速率限制模式用于控制资源使用
  6. 池化模式用于提高资源利用效率

掌握这些模式能够帮助我们写出更加高效、可靠的并发程序。在实际应用中，我们常常需要将这些基本模式组合使用，以满足特定的业务需求。

更新于 2025-02-11 
CC BY-NC-SA 4.0
GolangConcurrency
返回 | 主页
0%