golang

Channel 模式：掌握 Go 并发编程的精髓

深入学习 Go 的各种 Channel 使用模式，包括 Pipeline、Fan-in/Fan-out、Worker Pool 等经典并发模式

Leeting Yan 9 分钟阅读 4492 字

Channel 模式：掌握 Go 并发编程的精髓

Go 语言的 Channel 是并发编程的核心，但仅仅知道如何创建和使用 Channel 是不够的。掌握各种 Channel 模式（Channel Patterns）才能让你写出优雅、高效、可维护的并发程序。

本文将深入介绍 Go 中最常用的 Channel 模式，通过实战案例帮你掌握并发编程的精髓。

基础回顾

Channel 的基本操作

// 创建 channel
ch := make(chan int)        // 无缓冲
ch := make(chan int, 10)    // 缓冲容量为 10

// 发送和接收
ch <- 42        // 发送
value := <-ch   // 接收

// 关闭 channel
close(ch)

// 检查 channel 是否关闭
value, ok := <-ch
if !ok {
    // channel 已关闭
}

黄金法则

不要通过共享内存来通信，而要通过通信来共享内存。

Channel 底层实现与内存模型

理解 Channel 的底层机制有助于编写高性能并发代码。

数据结构

// runtime 中的 hchan 结构（简化版）
type hchan struct {
    qcount   uint           // 当前队列中的元素数量
    dataqsiz uint           // 环形队列的大小（缓冲容量）
    buf      unsafe.Pointer // 指向环形队列的指针
    elemsize uint16         // 元素大小
    closed   uint32         // 是否已关闭
    elemtype *_type         // 元素类型
    sendx    uint           // 发送索引
    recvx    uint           // 接收索引
    recvq    waitq          // 等待接收的 goroutine 队列
    sendq    waitq          // 等待发送的 goroutine 队列
    lock     mutex          // 互斥锁
}

关键洞察：

  • 无缓冲 channel 的 buf 为 nil，sendq 和 recvq 直接配对 goroutine
  • 缓冲 channel 使用环形队列，sendx 和 recvx 指针循环移动
  • channel 操作需要获取 lock，高并发场景下锁竞争是主要瓶颈

Happens-Before 关系

Go 内存模型对 channel 操作建立了 happens-before 保证：

操作Happens-Before 保证
ch <- v对 v 的写 happens-before 对应的 <-ch 读
close(ch)happens-before 任何返回 ok=false 的接收
带缓冲的 ch <- v如果接收方已准备好，同上；否则仅保证在队列中可见

这意味着 channel 不仅是数据传输管道，还是 goroutine 之间的内存同步原语。通过 channel 发送的数据，接收方一定能看到发送前对数据的所有修改。

select 的公平性

select 语句在多个 case 就绪时，会伪随机选择一个执行：

// 伪随机选择：不会永远优先第一个 case
for {
    select {
    case v := <-ch1:
        fmt.Println("ch1:", v)
    case v := <-ch2:
        fmt.Println("ch2:", v)
    }
}

注意：公平性是按 select 执行时的随机，而不是长期统计均匀。如果需要严格的轮询，应自行计数器控制。

模式 1：Pipeline（管道）

Pipeline 模式将处理流程分解为多个阶段，每个阶段通过 Channel 连接。

基本 Pipeline

package main

import (
    "fmt"
)

// 阶段 1：生成数字
func generate(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 filterEven(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            if n%2 == 0 {
                out <- n
            }
        }
    }()
    return out
}

func main() {
    // 构建 pipeline：生成 -> 平方 -> 过滤偶数
    nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squares := square(nums)
    evens := filterEven(squares)
    
    // 消费结果
    for result := range evens {
        fmt.Println(result)  // 4, 16, 36, 64, 100
    }
}

带错误处理的 Pipeline

package main

import (
    "errors"
    "fmt"
)

type Result struct {
    Value int
    Err   error
}

// 阶段 1：解析
func parse(inputs ...string) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for _, input := range inputs {
            var n int
            _, err := fmt.Sscanf(input, "%d", &n)
            if err != nil {
                out <- Result{Err: fmt.Errorf("parse %q: %w", input, err)}
                continue
            }
            out <- Result{Value: n}
        }
    }()
    return out
}

// 阶段 2：验证
func validate(in <-chan Result) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for r := range in {
            if r.Err != nil {
                out <- r
                continue
            }
            if r.Value < 0 {
                out <- Result{Err: errors.New("negative number")}
                continue
            }
            out <- r
        }
    }()
    return out
}

// 阶段 3：处理
func process(in <-chan Result) <-chan Result {
    out := make(chan Result)
    go func() {
        defer close(out)
        for r := range in {
            if r.Err != nil {
                out <- r
                continue
            }
            out <- Result{Value: r.Value * 2}
        }
    }()
    return out
}

func main() {
    inputs := []string{"1", "2", "abc", "-3", "4"}
    
    results := process(validate(parse(inputs...)))
    
    for r := range results {
        if r.Err != nil {
            fmt.Printf("Error: %v\n", r.Err)
        } else {
            fmt.Printf("Result: %d\n", r.Value)
        }
    }
}

模式 2：Fan-out/Fan-in（扇出/扇入）

Fan-out 将工作分发给多个 worker，Fan-in 将多个 worker 的结果合并。

Fan-out：分发工作

package main

import (
    "fmt"
    "sync"
)

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job)
        result := job * 2
        results <- result
    }
}

func main() {
    jobs := make(chan int, 100)
    results := make(chan int, 100)
    
    // Fan-out：启动 5 个 worker
    var wg sync.WaitGroup
    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            worker(id, jobs, results)
        }(i)
    }
    
    // 发送工作
    for i := 1; i <= 20; i++ {
        jobs <- i
    }
    close(jobs)
    
    // 等待所有 worker 完成
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // 收集结果
    for result := range results {
        fmt.Printf("Result: %d\n", result)
    }
}

Fan-in：合并结果

package main

import (
    "fmt"
    "sync"
)

// Fan-in：合并多个 channel
func merge(channels ...<-chan int) <-chan int {
    var wg sync.WaitGroup
    out := make(chan int)
    
    // 为每个输入 channel 启动一个 goroutine
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }
    
    // 等待所有 goroutine 完成后关闭输出 channel
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

func producer(id int, count int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for i := 0; i < count; i++ {
            out <- id*100 + i
        }
    }()
    return out
}

func main() {
    // 创建 3 个 producer
    p1 := producer(1, 5)
    p2 := producer(2, 5)
    p3 := producer(3, 5)
    
    // Fan-in：合并所有 producer 的输出
    merged := merge(p1, p2, p3)
    
    // 消费合并后的结果
    for v := range merged {
        fmt.Println(v)
    }
}

完整的 Fan-out/Fan-in 示例

package main

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

type Job struct {
    ID      int
    Payload string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

func worker(id int, jobs <-chan Job, results chan<- Result) {
    for job := range jobs {
        // 模拟处理时间
        time.Sleep(100 * time.Millisecond)
        
        result := Result{
            JobID:  job.ID,
            Output: fmt.Sprintf("Processed: %s", job.Payload),
        }
        results <- result
    }
}

func fanOut(jobs <-chan Job, workerCount int) []<-chan Result {
    channels := make([]<-chan Result, workerCount)
    
    for i := 0; i < workerCount; i++ {
        ch := make(chan Result)
        channels[i] = ch
        go worker(i, jobs, ch)
    }
    
    return channels
}

func fanIn(channels ...<-chan Result) <-chan Result {
    var wg sync.WaitGroup
    out := make(chan Result)
    
    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan Result) {
            defer wg.Done()
            for r := range c {
                out <- r
            }
        }(ch)
    }
    
    go func() {
        wg.Wait()
        close(out)
    }()
    
    return out
}

func main() {
    jobs := make(chan Job, 100)
    
    // 发送工作
    go func() {
        defer close(jobs)
        for i := 1; i <= 20; i++ {
            jobs <- Job{ID: i, Payload: fmt.Sprintf("Task %d", i)}
        }
    }()
    
    // Fan-out：分发给 5 个 worker
    resultChannels := fanOut(jobs, 5)
    
    // Fan-in：合并结果
    results := fanIn(resultChannels...)
    
    // 消费结果
    for result := range results {
        if result.Err != nil {
            fmt.Printf("Job %d failed: %v\n", result.JobID, result.Err)
        } else {
            fmt.Printf("Job %d: %s\n", result.JobID, result.Output)
        }
    }
}

模式 3：Worker Pool（工作池）

Worker Pool 模式创建固定数量的 worker 来处理任务。

基本 Worker Pool

package main

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

type WorkerPool struct {
    workerCount int
    jobs        chan Job
    results     chan Result
    wg          sync.WaitGroup
}

func NewWorkerPool(workerCount int) *WorkerPool {
    return &WorkerPool{
        workerCount: workerCount,
        jobs:        make(chan Job, 100),
        results:     make(chan Result, 100),
    }
}

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

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    
    for job := range wp.jobs {
        fmt.Printf("Worker %d started job %d\n", id, job.ID)
        time.Sleep(time.Second)  // 模拟工作
        
        result := Result{
            JobID:  job.ID,
            Output: fmt.Sprintf("Result of job %d", job.ID),
        }
        wp.results <- result
        fmt.Printf("Worker %d finished job %d\n", id, job.ID)
    }
}

func (wp *WorkerPool) Submit(job Job) {
    wp.jobs <- job
}

func (wp *WorkerPool) Results() <-chan Result {
    return wp.results
}

func (wp *WorkerPool) Shutdown() {
    close(wp.jobs)
    wp.wg.Wait()
    close(wp.results)
}

func main() {
    pool := NewWorkerPool(3)
    pool.Start()
    
    // 提交任务
    for i := 1; i <= 10; i++ {
        pool.Submit(Job{ID: i, Payload: fmt.Sprintf("Task %d", i)})
    }
    
    // 收集结果
    go func() {
        for result := range pool.Results() {
            fmt.Printf("Received: %s\n", result.Output)
        }
    }()
    
    // 等待完成
    pool.Shutdown()
}

带超时的 Worker Pool

package main

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

type TimeoutWorkerPool struct {
    workerCount int
    timeout     time.Duration
    jobs        chan Job
    results     chan Result
    wg          sync.WaitGroup
}

func (wp *TimeoutWorkerPool) worker(id int) {
    defer wp.wg.Done()
    
    for job := range wp.jobs {
        ctx, cancel := context.WithTimeout(context.Background(), wp.timeout)
        
        result := make(chan Result, 1)
        go func() {
            // 模拟工作
            time.Sleep(time.Duration(job.ID) * 500 * time.Millisecond)
            result <- Result{
                JobID:  job.ID,
                Output: fmt.Sprintf("Completed job %d", job.ID),
            }
        }()
        
        select {
        case r := <-result:
            wp.results <- r
        case <-ctx.Done():
            wp.results <- Result{
                JobID: job.ID,
                Err:   fmt.Errorf("timeout"),
            }
        }
        
        cancel()
    }
}

模式 4：Generator（生成器）

Generator 模式使用 Channel 实现惰性求值。

基本 Generator

package main

import (
    "fmt"
)

// 生成斐波那契数列
func fibonacci() <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        a, b := 0, 1
        for {
            ch <- a
            a, b = b, a+b
        }
    }()
    return ch
}

func main() {
    fib := fibonacci()
    
    // 只取前 10 个
    for i := 0; i < 10; i++ {
        fmt.Println(<-fib)
    }
}

带停止信号的 Generator

package main

import (
    "fmt"
)

func counter(stop <-chan struct{}) <-chan int {
    ch := make(chan int)
    go func() {
        defer close(ch)
        i := 0
        for {
            select {
            case <-stop:
                return
            case ch <- i:
                i++
            }
        }
    }()
    return ch
}

func main() {
    stop := make(chan struct{})
    count := counter(stop)
    
    // 取前 5 个
    for i := 0; i < 5; i++ {
        fmt.Println(<-count)
    }
    
    // 停止生成器
    close(stop)
}

模式 5：Or-Done Channel

Or-Done 模式允许在多个 channel 中任意一个关闭时触发操作。

package main

import (
    "fmt"
    "time"
)

// orDone 返回一个 channel，当任意输入 channel 关闭时关闭
func orDone(channels ...<-chan struct{}) <-chan struct{} {
    switch len(channels) {
    case 0:
        return nil
    case 1:
        return channels[0]
    }
    
    orDone := make(chan struct{})
    go func() {
        defer close(orDone)
        
        select {
        case <-channels[0]:
        case <-channels[1]:
        case <-orDone(channels[2:]...):
        }
    }()
    
    return orDone
}

func main() {
    sig := func(after time.Duration) <-chan struct{} {
        c := make(chan struct{})
        go func() {
            defer close(c)
            time.Sleep(after)
        }()
        return c
    }
    
    start := time.Now()
    
    <-orDone(
        sig(2*time.Hour),
        sig(5*time.Minute),
        sig(1*time.Second),  // 这个会最先触发
        sig(1*time.Hour),
    )
    
    fmt.Printf("Done after: %v\n", time.Since(start))
}

模式 6：Tee Channel（分流）

Tee 模式将一个 channel 的数据分流到两个 channel。

package main

import (
    "fmt"
)

func tee(in <-chan int) (<-chan int, <-chan int) {
    out1 := make(chan int)
    out2 := make(chan int)
    
    go func() {
        defer close(out1)
        defer close(out2)
        
        for val := range in {
            // 使用局部变量避免闭包问题
            val1, val2 := val, val
            
            // 并发发送到两个 channel
            select {
            case out1 <- val1:
            }
            select {
            case out2 <- val2:
            }
        }
    }()
    
    return out1, out2
}

func main() {
    // 创建数据源
    source := make(chan int)
    go func() {
        defer close(source)
        for i := 1; i <= 5; i++ {
            source <- i
        }
    }()
    
    // 分流
    ch1, ch2 := tee(source)
    
    // 消费两个 channel
    for i := 1; i <= 5; i++ {
        v1 := <-ch1
        v2 := <-ch2
        fmt.Printf("ch1: %d, ch2: %d\n", v1, v2)
    }
}

模式 7：Bridge Channel（桥接）

Bridge 模式将多个 channel 串联成一个。

package main

import (
    "fmt"
)

func bridge(chanStream <-chan <-chan int) <-chan int {
    out := make(chan int)
    
    go func() {
        defer close(out)
        
        for ch := range chanStream {
            for val := range ch {
                out <- val
            }
        }
    }()
    
    return out
}

func main() {
    // 创建 channel 的 stream
    chanStream := make(chan (<-chan int))
    
    go func() {
        defer close(chanStream)
        
        // 发送多个 channel
        for i := 0; i < 3; i++ {
            ch := make(chan int, 3)
            for j := 0; j < 3; j++ {
                ch <- i*10 + j
            }
            close(ch)
            chanStream <- ch
        }
    }()
    
    // 桥接所有 channel
    for val := range bridge(chanStream) {
        fmt.Println(val)
    }
}

模式 8：Rate Limiting（限流）

使用 Channel 实现速率限制。

简单限流

package main

import (
    "fmt"
    "time"
)

func main() {
    // 每秒最多处理 3 个请求
    limiter := time.Tick(333 * time.Millisecond)
    
    requests := make(chan int, 10)
    for i := 1; i <= 10; i++ {
        requests <- i
    }
    close(requests)
    
    for req := range requests {
        <-limiter  // 等待限流器
        fmt.Printf("Processing request %d at %v\n", req, time.Now())
    }
}

令牌桶限流

package main

import (
    "fmt"
    "time"
)

type RateLimiter struct {
    tokens chan struct{}
}

func NewRateLimiter(rate int, burst int) *RateLimiter {
    rl := &RateLimiter{
        tokens: make(chan struct{}, burst),
    }
    
    // 以固定速率添加令牌
    go func() {
        ticker := time.NewTicker(time.Second / time.Duration(rate))
        defer ticker.Stop()
        
        for range ticker.C {
            select {
            case rl.tokens <- struct{}{}:
            default:
                // 桶已满，丢弃令牌
            }
        }
    }()
    
    return rl
}

func (rl *RateLimiter) Wait() {
    <-rl.tokens
}

func main() {
    limiter := NewRateLimiter(5, 10)  // 每秒 5 个请求，突发 10 个
    
    for i := 1; i <= 20; i++ {
        limiter.Wait()
        fmt.Printf("Request %d at %v\n", i, time.Now())
    }
}

模式 9：Cancellation（取消）

使用 Channel 实现优雅的取消机制。

package main

import (
    "fmt"
    "time"
)

func longRunningTask(cancel <-chan struct{}) {
    for i := 0; i < 10; i++ {
        select {
        case <-cancel:
            fmt.Println("Task cancelled")
            return
        default:
            fmt.Printf("Working... %d\n", i)
            time.Sleep(500 * time.Millisecond)
        }
    }
    fmt.Println("Task completed")
}

func main() {
    cancel := make(chan struct{})
    
    go longRunningTask(cancel)
    
    // 2 秒后取消
    time.Sleep(2 * time.Second)
    close(cancel)
    
    time.Sleep(1 * time.Second)
}

总结

Channel 模式是 Go 并发编程的精髓：

核心模式：

  1. Pipeline：将处理流程分解为多个阶段
  2. Fan-out/Fan-in：分发和合并工作
  3. Worker Pool：固定数量的 worker 处理任务
  4. Generator：惰性求值
  5. Or-Done：多路选择
  6. Tee：分流数据
  7. Bridge：桥接多个 channel
  8. Rate Limiting：速率限制
  9. Cancellation：优雅取消

最佳实践：

  1. 总是关闭发送端的 channel
  2. 使用 select 处理多个 channel
  3. 使用 context 进行取消和超时控制
  4. 避免 goroutine 泄漏
  5. 使用缓冲 channel 提高性能

记住：并发不是目的，而是手段。选择正确的模式，让并发程序既高效又易于理解。

性能基准：Channel vs Mutex

什么时候用 channel，什么时候用 sync.Mutex？以下是基于 Go 1.22 的基准对比：

场景Channel (带缓冲)sync.Mutex + []T建议
单生产者单消费者~50 ns/op~30 ns/opMutex 稍快，但 channel 更安全
多生产者单消费者~80 ns/op~60 ns/op + 竞争Channel 避免手动锁管理
任务分发 (Worker Pool)~100 ns/op需额外实现Channel 更符合 Go 哲学
共享计数器N/A（不适合）~15 ns/opMutex 是正确选择
// 基准测试示例
func BenchmarkChanTransfer(b *testing.B) {
    ch := make(chan int, 1)
    go func() {
        for v := range ch {
            _ = v
        }
    }()
    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        ch <- i
    }
    close(ch)
}

规则：数据流动用 channel，状态保护用 mutex。不要把 channel 当 mutex 用（性能差且代码难读），也不要把 mutex 当 channel 用（容易遗漏 Unlock）。

常见死锁模式与预防

Go 运行时能检测部分死锁，但理解成因才能避免：

死锁模式示例预防
无缓冲 channel 双向等待A 等 B 发，B 等 A 发使用缓冲 channel 或额外 goroutine 协调
忘记关闭 channelrange 永远阻塞明确关闭责任方；或用 context 取消
锁顺序错误先锁 A 再锁 B vs 先锁 B 再锁 A全局统一加锁顺序
select 无 default 且无 case 就绪所有 goroutine 阻塞确保至少一个 case 最终会就绪
// 死锁示例：双向等待
func deadlockExample() {
    ch1 := make(chan int)
    ch2 := make(chan int)

    go func() {
        ch1 <- 1      // 阻塞在这里...
        <-ch2
    }()

    <-ch1             // 等 ch1 有数据
    ch2 <- 2          // 但上面的 goroutine 也在等 ch2 有容量
    // 死锁！
}

Go 运行时发现所有 goroutine 都阻塞时会 fatal error: all goroutines are asleep - deadlock!，但生产环境中更危险的是部分 goroutine 死锁（其他 goroutine 仍在运行，不会触发 panic）。

context 与 channel 的结合

context.Context 是现代 Go 程序的标准取消机制，与 channel 配合实现超时和级联取消：

func processWithTimeout(jobs <-chan Job) <-chan Result {
    results := make(chan Result)

    go func() {
        defer close(results)

        for job := range jobs {
            // 每个 job 最多执行 5 秒
            ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

            resultCh := make(chan Result, 1)
            go func(j Job) {
                resultCh <- processJob(j)
            }(job)

            select {
            case r := <-resultCh:
                results <- r
            case <-ctx.Done():
                results <- Result{Err: ctx.Err()}  // context deadline exceeded
            }

            cancel()
        }
    }()

    return results
}
模式适用场景
context.WithTimeout单个操作超时（如 HTTP 请求）
context.WithDeadline截止时间控制（如整点结算）
context.WithCancel手动取消（如用户中断）
<-ctx.Done()在 select 中与业务 channel 并列，实现优雅退出

高级：生产者消费者扩展

背压控制（Backpressure）

当消费者速度慢于生产者时，无限制缓冲会导致内存爆炸。背压控制让生产者感知消费者压力：

func producerWithBackpressure(consumerSpeed time.Duration) <-chan int {
    // 故意使用小缓冲，让发送方在消费者慢时阻塞
    out := make(chan int, 2)

    go func() {
        defer close(out)
        for i := 0; ; i++ {
            select {
            case out <- i:
                // 成功发送
            case <-time.After(100 * time.Millisecond):
                // 缓冲满，消费者跟不上——降级逻辑
                fmt.Println("Consumer lagging, skipping", i)
            }
        }
    }()

    return out
}

动态 Worker Pool

根据负载自动扩缩容 worker 数量：

type DynamicPool struct {
    minWorkers, maxWorkers int
    jobs                   chan Job
    results                chan Result
    wg                     sync.WaitGroup
    active                 int32  // atomic counter
}

func (p *DynamicPool) scale() {
    load := len(p.jobs)
    active := int(atomic.LoadInt32(&p.active))

    if load > active*2 && active < p.maxWorkers {
        // 增加 worker
        p.wg.Add(1)
        atomic.AddInt32(&p.active, 1)
        go func() {
            defer p.wg.Done()
            defer atomic.AddInt32(&p.active, -1)
            p.worker()
        }()
    }
}

总结：Channel 不是 Go 并发的银弹，但它是最符合 Go 语言哲学的并发原语。理解其底层实现（环形队列 + 锁）、内存模型（happens-before）、以及何时与 sync 包工具配合，才能写出真正健壮的并发程序。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「golang」更多文章

  1. 《Go 语言编程实战》目录
  2. 《Go 语言编程实战》18.3 上线、观测与迭代
  3. 《Go 语言编程实战》18.2 故障演练