Go channel 深度实战——有缓冲/无缓冲、select 多路复用、pipeline 模式

老张2026/4/30大约 8 分钟

Go channel 深度实战——有缓冲/无缓冲、select 多路复用、pipeline 模式

适读人群：有Go基础、想真正用好并发通信的工程师 | 阅读时长：约18分钟 | 核心价值：channel不只是队列，是Go并发设计哲学的核心载体

一、小林的「消息队列」事故

小林在一家电商公司负责订单服务，技术栈从Java迁到Go大概半年。有一次需求是：订单创建成功后，异步触发库存扣减、积分发放、消息通知三个操作，三个操作互相独立，但要等全部完成后才给前端返回结果。

他用Java思维写了这样的代码：把三个操作都通过channel发出去，然后等返回：

// 小林的原始版本（有问题）
func processOrder(order Order) error {
    ch := make(chan error)

    go func() { ch <- deductInventory(order) }()
    go func() { ch <- grantPoints(order) }()
    go func() { ch <- sendNotification(order) }()

    // 等待3个结果
    for i := 0; i < 3; i++ {
        if err := <-ch; err != nil {
            return err  // 这里有问题！
        }
    }
    return nil
}

代码看起来没毛病，但压测时偶发goroutine泄漏。原因：当第一个操作返回错误时，函数提前return了，但另外两个goroutine还在往 ch 发送数据，因为 ch 是无缓冲channel，没人接收，它们永久阻塞。

这是个经典的channel使用误区。今天我把channel相关的坑和正确用法系统梳理一遍。

二、无缓冲 vs 有缓冲：不只是性能差异

无缓冲channel（同步通信）

ch := make(chan int) // 无缓冲

发送和接收必须同时ready，才能完成一次通信。本质上是同步握手：

发送方会阻塞，直到接收方准备好
接收方会阻塞，直到发送方发来数据

适合场景： 同步信号、确保对方已经收到、goroutine协调

有缓冲channel（异步通信）

ch := make(chan int, 10) // 缓冲大小10

发送方只要缓冲区未满就不阻塞，接收方只要缓冲区非空就不阻塞。本质上是异步队列：

发送方满了才阻塞
接收方空了才阻塞

适合场景： 生产者消费者、削峰填谷、worker pool任务分发

package main

import (
    "fmt"
    "time"
)

func unbufferedDemo() {
    ch := make(chan string)

    go func() {
        fmt.Println("发送方：准备发送")
        ch <- "hello"  // 阻塞直到接收方ready
        fmt.Println("发送方：发送完成")
    }()

    time.Sleep(100 * time.Millisecond) // 模拟接收方延迟
    fmt.Println("接收方：准备接收")
    msg := <-ch
    fmt.Println("接收方：收到", msg)
}

func bufferedDemo() {
    ch := make(chan string, 3) // 缓冲3个

    // 发送方可以连发3个而不阻塞
    ch <- "msg1"
    ch <- "msg2"
    ch <- "msg3"
    fmt.Println("发送了3条消息，缓冲区满了")
    // ch <- "msg4"  // 这里会阻塞，因为缓冲区满了

    fmt.Println(<-ch) // msg1
    fmt.Println(<-ch) // msg2
    fmt.Println(<-ch) // msg3
}

func main() {
    fmt.Println("=== 无缓冲 ===")
    unbufferedDemo()
    fmt.Println("=== 有缓冲 ===")
    bufferedDemo()
}

三、channel的方向类型：只读和只写

Go允许把channel限定方向，这是接口设计的好工具，让代码意图更清晰：

package main

import "fmt"

// 只接收参数（消费者）
func consumer(ch <-chan int) {
    for v := range ch {
        fmt.Println("消费:", v)
    }
}

// 只发送参数（生产者）
func producer(ch chan<- int) {
    for i := 0; i < 5; i++ {
        ch <- i
    }
    close(ch) // 生产者负责关闭
}

func main() {
    ch := make(chan int, 5)
    go producer(ch)  // 双向channel可以传给单向参数
    consumer(ch)
}

四、select 多路复用：Go并发控制的瑞士军刀

select 是Go处理多个channel的核心语法，类似于网络编程里的 select/epoll：

select {
case v := <-ch1:
    // ch1有数据
case v := <-ch2:
    // ch2有数据
case ch3 <- val:
    // 成功发送到ch3
default:
    // 所有case都未就绪时执行
}

关键特性：

多个case同时ready时，随机选一个执行（公平调度）
没有ready的case且有default：立即执行default（非阻塞）
没有ready的case且没有default：阻塞等待

实战用法1：超时控制

package main

import (
    "fmt"
    "time"
)

func slowOperation() <-chan string {
    ch := make(chan string, 1)
    go func() {
        time.Sleep(2 * time.Second) // 模拟慢操作
        ch <- "操作完成"
    }()
    return ch
}

func withTimeout() {
    resultCh := slowOperation()

    select {
    case result := <-resultCh:
        fmt.Println("成功:", result)
    case <-time.After(1 * time.Second): // 1秒超时
        fmt.Println("操作超时！")
    }
}

func main() {
    withTimeout()
}

实战用法2：退出信号

package main

import (
    "fmt"
    "time"
)

func worker(done <-chan struct{}) {
    for {
        select {
        case <-done:
            fmt.Println("worker退出")
            return
        default:
            fmt.Println("worker工作中...")
            time.Sleep(300 * time.Millisecond)
        }
    }
}

func main() {
    done := make(chan struct{})

    go worker(done)

    time.Sleep(1 * time.Second)
    close(done) // 通知worker退出，close会广播给所有接收方
    time.Sleep(100 * time.Millisecond)
    fmt.Println("主程序结束")
}

实战用法3：非阻塞发送/接收

package main

import "fmt"

func tryReceive(ch <-chan int) (int, bool) {
    select {
    case v := <-ch:
        return v, true
    default:
        return 0, false
    }
}

func trySend(ch chan<- int, val int) bool {
    select {
    case ch <- val:
        return true
    default:
        return false // 发送失败（缓冲区满）
    }
}

func main() {
    ch := make(chan int, 2)
    ch <- 1

    if v, ok := tryReceive(ch); ok {
        fmt.Println("收到:", v)
    }
    if _, ok := tryReceive(ch); !ok {
        fmt.Println("channel空了")
    }

    fmt.Println("发送成功:", trySend(ch, 10))
    fmt.Println("发送成功:", trySend(ch, 20))
    fmt.Println("发送成功:", trySend(ch, 30)) // false，缓冲区满
}

五、pipeline模式：用channel串联数据流

pipeline是Go并发的经典模式，把数据处理分成多个stage，每个stage通过channel连接，形成流水线。这和Java里的Stream API有点像，但更底层、更灵活。

package main

import (
    "fmt"
    "math"
)

// Stage 1: 生成自然数
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 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
}

// Stage 3: 过滤（只保留完全平方数的平方根为整数的）
func filterPerfectSquares(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            sqrt := math.Sqrt(float64(n))
            if sqrt == math.Floor(sqrt) {
                out <- n
            }
        }
    }()
    return out
}

func main() {
    // 串联pipeline：生成 → 平方 → 过滤
    nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    squares := square(nums)
    results := filterPerfectSquares(squares)

    for v := range results {
        fmt.Println(v) // 1, 4, 9, 16, 25, 36, 49, 64, 81, 100
    }
}

fan-out / fan-in：扇出合并模式

package main

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

// fan-out：一个输入channel，分发给多个worker
func fanOut(in <-chan int, workerCount int) []<-chan int {
    outs := make([]<-chan int, workerCount)
    for i := 0; i < workerCount; i++ {
        outs[i] = worker(in, i)
    }
    return outs
}

func worker(in <-chan int, id int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            time.Sleep(10 * time.Millisecond) // 模拟处理耗时
            out <- n * n
            fmt.Printf("worker%d 处理了 %d\n", id, n)
        }
    }()
    return out
}

// fan-in：多个输入channel，合并到一个输出channel
func fanIn(ins ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, in := range ins {
        wg.Add(1)
        go func(ch <-chan int) {
            defer wg.Done()
            for v := range ch {
                out <- v
            }
        }(in)
    }

    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

func main() {
    in := make(chan int, 10)
    for i := 1; i <= 6; i++ {
        in <- i
    }
    close(in)

    // 3个worker并行处理
    outs := fanOut(in, 3)

    // 合并结果
    merged := fanIn(outs...)
    for v := range merged {
        fmt.Println("结果:", v)
    }
}

六、channel使用的三大陷阱

陷阱1：向已关闭的channel发送数据会panic

package main

import "fmt"

func main() {
    ch := make(chan int, 3)
    close(ch)

    // panic: send on closed channel
    // ch <- 1

    // 接收已关闭的channel：立即返回零值和false
    v, ok := <-ch
    fmt.Println(v, ok) // 0 false

    // range会自动停止
    ch2 := make(chan int, 3)
    ch2 <- 1
    ch2 <- 2
    close(ch2)
    for v := range ch2 {
        fmt.Println(v) // 1, 2（遇到close自动退出）
    }
}

原则：只有发送方才能关闭channel，接收方不要关闭。

陷阱2：关闭nil channel会panic

package main

func main() {
    var ch chan int // nil channel

    // panic: close of nil channel
    // close(ch)

    // 从nil channel接收：永久阻塞（不是panic）
    // <-ch

    // 往nil channel发送：永久阻塞
    // ch <- 1
}

原则：channel使用前必须初始化（make），nil channel只在特定技巧中使用。

陷阱3：回到小林的问题——提前return导致goroutine泄漏

package main

import (
    "errors"
    "fmt"
    "sync"
)

// 正确修复：用带缓冲的channel，或者用errgroup
func processOrderCorrect(order string) error {
    // 方案1：有缓冲channel，发送方不会阻塞
    errCh := make(chan error, 3) // 缓冲大小 = goroutine数量

    go func() { errCh <- deductInventory(order) }()
    go func() { errCh <- grantPoints(order) }()
    go func() { errCh <- sendNotification(order) }()

    var firstErr error
    for i := 0; i < 3; i++ {
        if err := <-errCh; err != nil && firstErr == nil {
            firstErr = err
            // 不提前return，让3个goroutine都能完成发送
        }
    }
    return firstErr
}

// 方案2：用errgroup（推荐）
func processOrderErrGroup(order string) error {
    var wg sync.WaitGroup
    errs := make([]error, 3)

    tasks := []func() error{
        func() error { return deductInventory(order) },
        func() error { return grantPoints(order) },
        func() error { return sendNotification(order) },
    }

    for i, task := range tasks {
        wg.Add(1)
        i, task := i, task
        go func() {
            defer wg.Done()
            errs[i] = task()
        }()
    }
    wg.Wait()

    for _, err := range errs {
        if err != nil {
            return err
        }
    }
    return nil
}

func deductInventory(order string) error { return nil }
func grantPoints(order string) error     { return errors.New("积分服务异常") }
func sendNotification(order string) error { return nil }

func main() {
    err := processOrderCorrect("order-001")
    fmt.Println("方案1结果:", err)

    err = processOrderErrGroup("order-002")
    fmt.Println("方案2结果:", err)
}

七、Java BlockingQueue vs Go channel

Java工程师用BlockingQueue实现生产者消费者，Go的channel是内置的同等抽象：

Java BlockingQueue	Go channel
`ArrayBlockingQueue(n)`	`make(chan T, n)`
`LinkedBlockingQueue()`	无直接等价（可用无限缓冲方案）
`put()`（满了阻塞）	`ch <- v`（满了阻塞）
`take()`（空了阻塞）	`<-ch`（空了阻塞）
`offer()`（非阻塞）	select + default
`poison pill模式`	`close(ch)`

八、总结

channel的本质是有类型的并发通信管道，而不只是线程安全队列。用好channel需要掌握：

无缓冲 = 同步握手，有缓冲 = 异步队列，根据场景选择
select是多路复用核心，实现超时、退出信号、非阻塞操作
pipeline = 数据流水线，让并发处理清晰可读
关闭原则：只有发送方关闭，关闭即广播，nil channel要避免
缓冲大小 = goroutine数量，防止提前return导致阻塞泄漏

Go的设计哲学：「不要通过共享内存来通信，而要通过通信来共享内存。」channel是这个哲学的具体实现。