Go · Channel 与 select

场景 1 · worker pool 控并发保护下游

批量导出 10 万单, 下游 DB 只能扛 20 并发:

jobs := make(chan Order, 1000)        // 有缓冲: 吸收突发, 反压生产者
results := make(chan Result, 1000)
var wg sync.WaitGroup
for i := 0; i < 20; i++ {              // worker 数 = 下游能扛的并发
    wg.Add(1)
    go func() {
        defer wg.Done()
        for order := range jobs {         // close(jobs) 后自动退出
            results <- process(order)
        }
    }()
}
go func() { for _, o := range allOrders { jobs <- o }; close(jobs) }()
wg.Wait(); close(results)                  // 消费者 range results 优雅结束

场景 2 · 每个下游调用都套超时

select {
case v := <-queryCh:
    return v
case <-time.After(800 * time.Millisecond):
    return ErrTimeout                   // 熔断慢查询, 保住整体延迟
case <-ctx.Done():                         // 用户取消/整链超时
    return ctx.Err()
}

场景 3 · 扇入聚合多个数据源

func merge(ctx context.Context, chans ...<-chan Item) <-chan Item {
    out := make(chan Item)
    var wg sync.WaitGroup
    wg.Add(len(chans))
    for _, c := range chans {
        go func(c <-chan Item) {              // 每个来源一个搬运 goroutine
            defer wg.Done()
            for it := range c {
                select { case out <- it: case <-ctx.Done(): return }
            }
        }(c)
    }
    go func() { wg.Wait(); close(out) }()
    return out
}

场景 4 · 令牌桶限流 channel 化

Ticker 定期投放令牌, 缓冲即容量; 拿不到令牌的请求在接收处排队:

func limiter(rate int, burst int) <-chan struct{} {
    tokens := make(chan struct{}, burst)      // 缓冲 = 桶容量
    go func() {
        t := time.NewTicker(time.Second / time.Duration(rate))
        for range t.C {
            select { case tokens <- struct{}{}: default: }  // 满则丢弃, 不阻塞泵
        }
    }()
    return tokens
}
<-limiter(100, 50)                          // 调用点: 拿到令牌才放行

场景 5 · fan-out + 首错取消

并发打 N 个请求, 第一个失败就取消其余, 不浪费下游配额:

func fetchAll(ctx context.Context, urls []string) ([]Result, error) {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()                            // 兜底: 正常返回也释放
    errs := make(chan error, 1)           // 缓冲1: 首个错误不阻塞报告者
    results := make([]Result, len(urls))
    var wg sync.WaitGroup
    for i, u := range urls {
        wg.Add(1)
        go func(i int, u string) {
            defer wg.Done()
            r, err := fetchOne(ctx, u)       // ctx 取消时快速失败返回
            if err != nil {
                select { case errs <- err: cancel(): default: }  // 只报第一个错
                return
            }
            results[i] = r                    // 各写各下标: 无竞争
        }(i, u)
    }
    wg.Wait()
    select { case err := <-errs: return nil, err: default: return results, nil }
}

场景 6 · 事件订阅: 广播与退出

发布订阅用 chan 切片 + close 广播; 订阅者 range 即自动退出:

type Bus struct {
    mu   sync.Mutex
    subs map[chan Event]struct{}         // 每订阅者一条 chan
}
func (b *Bus) Subscribe() <-chan Event {
    ch := make(chan Event, 16)         // 小缓冲吸收突发
    b.mu.Lock(); b.subs[ch] = struct{}{}; b.mu.Unlock()
    return ch
}
func (b *Bus) Publish(e Event) {
    b.mu.Lock(); defer b.mu.Unlock()
    for ch := range b.subs {
        select { case ch <- e: default: }     // 满则丢: 慢订阅者不拖垮发布
    }
}

场景 7 · 流水线 pipeline(生成器→阶段→汇)

经典三段式: 源生成、阶段变换、汇聚消费, 每阶段一个 goroutine:

func gen(ctx context.Context, nums ...int) <-chan int {
    out := make(chan int)
    go func() { defer close(out)
        for _, n := range nums { select { case out <- n: case <-ctx.Done(): return } }
    }()
    return out
}
func sq(in <-chan int) <-chan int {         // 阶段: 变换
    out := make(chan int)
    go func() { defer close(out)
        for n := range in { out <- n * n }   // 上游 close 后自动结束
    }()
    return out
}
// 消费: for v := range sq(gen(ctx, 1, 2, 3)) { ... }

场景 8 · 心跳通道: 判活与慢消费者检测

处理循环同时监听心跳与数据, 心跳超时说明上游卡死:

heartbeat := time.NewTicker(5 * time.Second)
defer heartbeat.Stop()
for {
    select {
    case item := <-dataCh:
        process(item)                      // 正常路径
    case <-heartbeat.C:                    // 5s 没数据也醒来: 记活/上报
        markAlive()
    case <-ctx.Done():
        return
    }
}

场景 9 · 优雅停机三件套

ctx 取消 + WaitGroup 等待 + done 通知, 三者配合不丢任务:

ctx, cancel := context.WithCancel(context.Background())
go handleSignals(cancel)                  // SIGTERM → cancel()

<-ctx.Done()                              // 1. 收到停止信号
server.SetKeepAlivesEnabled(false)         // 2. 不再接新连接
ctxShutdown, c2 := context.WithTimeout(context.Background(), 15*time.Second)
defer c2()
server.Shutdown(ctxShutdown)              // 3. 等存量请求完成, 超时强制断

场景 10 · 每请求超时贯穿到 DAO

超时不该只在 HTTP 层 —— ctx 一路传下去, 慢 SQL 一起被掐断:

func (h *Handler) GetOrder(w http.ResponseWriter, r *http.Request) {
    ctx, cancel := context.WithTimeout(r.Context(), 800*time.Millisecond)
    defer cancel()
    order, err := h.dao.GetOrder(ctx, id)     // dao 用 ctx 执行查询
    if errors.Is(err, context.DeadlineExceeded) {
        http.Error(w, "timeout", http.StatusGatewayTimeout)
        return                                  // 上层 800ms 处置, 不拖 worker
    }
    json.NewEncoder(w).Encode(order)
}