第四章：Channel高级模式：单向Channel、Select多路复用、超时控制与Ticker、Fan-in与Fan-out模式

Channel 这东西，说白了就是 Go 并发模型的灵魂。前面我们聊了基本的 channel 用法，但实际项目中，光会 ch <- 和 <-ch 可不够。今天咱们来点硬核的——那些真正能让你写出优雅并发代码的高级模式。

我个人习惯把 channel 的高级用法分成四块：单向 channel 的类型约束、select 的多路复用、超时控制与 ticker 的配合，以及经典的 fan-in/fan-out 模式。这四块掌握了，你基本就能应付 90% 的并发场景了。

4.1 单向Channel：给类型加上方向约束

先问个问题：你写的函数参数里，channel 是不是一直都是 chan T？嗯，我以前也是。直到有一次代码 review，同事指着我的函数签名说：「你这个 channel 到底是用来发的还是收的？」我才意识到，不加方向约束，代码的可读性和安全性都会打折扣。

单向 channel 其实很简单，就是在类型声明时加上 <- 箭头：

  • chan<- T：只写 channel，只能发送数据
  • <-chan T：只读 channel，只能接收数据

看个例子就明白了：

// 生产者：只能发送
func producer(out chan<- int) {
    for i := 0; i < 5; i++ {
        out <- i
    }
    close(out)
}

// 消费者：只能接收
func consumer(in <-chan int) {
    for val := range in {
        fmt.Println("消费:", val)
    }
}

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

我在项目中遇到过这样的坑：一个协程不小心往 channel 里写入了错误的数据，排查了半天才发现是函数内部误用了 channel 的方向。从那以后，我所有公开的 API 函数，只要涉及 channel 参数，一律加上方向约束。这不仅是文档，更是编译器级别的保护。

小技巧：双向 channel 可以隐式转换为单向 channel，但反过来不行。所以你在函数参数里用单向 channel，调用方传双向 channel 是没问题的。

4.2 Select多路复用：同时监听多个Channel

select 语句，我觉得是 Go 并发里最优雅的设计之一。它让你能同时等待多个 channel 操作，哪个先准备好就执行哪个。有点像 switch，但专门为 channel 设计的。

func main() {
    ch1 := make(chan string)
    ch2 := make(chan string)

    go func() {
        time.Sleep(1 * time.Second)
        ch1 <- "来自 ch1"
    }()
    go func() {
        time.Sleep(2 * time.Second)
        ch2 <- "来自 ch2"
    }()

    select {
    case msg := <-ch1:
        fmt.Println(msg)
    case msg := <-ch2:
        fmt.Println(msg)
    case <-time.After(500 * time.Millisecond):
        fmt.Println("超时了！")
    }
}

这里有个细节要注意：如果多个 case 同时满足，select 会随机选一个执行。这不是 bug，是故意设计的——防止你写出依赖顺序的代码。

我曾经在写一个网关服务时，需要同时监听多个上游服务的响应。用 select 配合多个 channel，代码清晰得不得了。要是用其他语言，估计得搞一堆回调或者事件循环。

注意：空的 select{} 会永久阻塞。如果你不小心写了个空的 select，程序就卡死了。我见过线上事故就是因为这个。

4.3 超时控制与Ticker：别让你的协程死等

说到超时控制，这是并发编程里最容易出问题的地方之一。你想想看，一个协程发了个请求，结果对方挂了，你的协程就永远卡在那了？这显然不行。

Go 里做超时控制，最常用的就是 time.After 和 time.Ticker。

4.3.1 time.After：一次性超时

func doWithTimeout() {
    ch := make(chan int)
    go func() {
        time.Sleep(3 * time.Second)
        ch <- 42
    }()

    select {
    case result := <-ch:
        fmt.Println("结果:", result)
    case <-time.After(2 * time.Second):
        fmt.Println("操作超时")
    }
}

这里 time.After 返回一个 channel，2 秒后它会收到一个值。配合 select，就能实现超时控制。简单吧？

4.3.2 time.Ticker：周期性任务

Ticker 适合做定时任务。比如每隔 5 秒检查一次服务健康状态：

func healthCheck() {
    ticker := time.NewTicker(5 * time.Second)
    defer ticker.Stop()

    for {
        select {
        case <-ticker.C:
            fmt.Println("执行健康检查...")
            // 检查逻辑
        case <-ctx.Done():
            fmt.Println("收到退出信号")
            return
        }
    }
}

我记得有一次，同事写了个定时任务，忘了调用 ticker.Stop()，结果 goroutine 泄漏了。嗯，这里要提醒一下：Ticker 用完后一定要 Stop，不然它会一直运行，GC 也回收不了。

核心要点：超时控制不是可选项，是必须项。任何可能阻塞的操作，都应该加上超时机制。这是生产级代码的基本素养。

4.4 Fan-in与Fan-out模式：并发编排的艺术

这两个模式，说白了就是「分而治之」和「合而为一」。

  • Fan-out：一个生产者，多个消费者。把任务分发给多个 worker 并行处理。
  • Fan-in：多个生产者，一个消费者。把多个 channel 的结果合并到一个 channel 里。

4.4.1 Fan-out：任务分发

func worker(id int, jobs <-chan int, results chan<- int) {
    for job := range jobs {
        fmt.Printf("Worker %d 处理任务 %d\n", id, job)
        time.Sleep(time.Second)
        results <- job * 2
    }
}

func main() {
    jobs := make(chan int, 10)
    results := make(chan int, 10)

    // 启动 3 个 worker（Fan-out）
    for w := 1; w <= 3; w++ {
        go worker(w, jobs, results)
    }

    // 发送 5 个任务
    for j := 1; j <= 5; j++ {
        jobs <- j
    }
    close(jobs)

    // 收集结果
    for r := 1; r <= 5; r++ {
        <-results
    }
}

4.4.2 Fan-in：结果合并

func fanIn(ch1, ch2 <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for {
            select {
            case v := <-ch1:
                out <- v
            case v := <-ch2:
                out <- v
            }
        }
    }()
    return out
}

我在做日志处理系统时，就用到了 fan-out/fan-in 模式。上游日志源不断产生数据，我 fan-out 给 10 个 worker 并行处理（解析、过滤、格式化），然后再 fan-in 到一个 channel 统一写入存储。吞吐量直接提升了 8 倍。

避坑指南：我曾经在 fan-in 时忘了处理 channel 关闭的情况，导致死循环。正确的做法是在 select 里检查 channel 是否关闭：case v, ok := <-ch1: if !ok { ... }

4.5 综合实战：一个完整的并发流水线

来，咱们把今天学的串起来，写一个完整的例子：

func main() {
    // 阶段1：生成数据
    nums := make(chan int)
    go func() {
        for i := 1; i <= 10; i++ {
            nums <- i
        }
        close(nums)
    }()

    // 阶段2：Fan-out 到 3 个 worker 处理
    results := make(chan int, 10)
    var wg sync.WaitGroup
    for i := 0; i < 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for n := range nums {
                time.Sleep(100 * time.Millisecond)
                results <- n * n
            }
        }(i)
    }

    // 阶段3：关闭 results（Fan-in 的收尾）
    go func() {
        wg.Wait()
        close(results)
    }()

    // 阶段4：消费结果，带超时控制
    for {
        select {
        case val, ok := <-results:
            if !ok {
                fmt.Println("所有任务完成")
                return
            }
            fmt.Println("结果:", val)
        case <-time.After(2 * time.Second):
            fmt.Println("处理超时")
            return
        }
    }
}

这个例子包含了：单向 channel（虽然没有显式声明，但你可以加上）、select 多路复用、超时控制、fan-out/fan-in。一个流水线，四种模式全用上了。

最后说一句：这些模式不是银弹。别为了用而用。如果你的场景就两个协程通信，一个 channel 就够了。但一旦涉及到多个生产者、多个消费者、超时控制、任务分发，这些高级模式就是你的利器。

嗯，今天就到这。下一章咱们聊聊 Context——Go 并发编程里另一个重量级角色。

← 上一章 📖 返回目录 下一章 →