English 中文 العربية 日本語 Português

Go: Go Select 与并发模式

select 是 Go 并发编程的瑞士军刀——同时监听多个 channel，谁先就绪处理谁。它是连接 goroutine 的"智能交换机"。

如果说 channel 是 goroutine 之间的电话线，select 就是接线员——它同时监听所有电话，哪个响就接哪个。这节课你将掌握 select 的全部核心模式。

1. 你将学到

2. 一个量化交易工程师的真实故事

(1) 痛点：轮询 3 个数据源，CPU 跑满

Bob 是量化交易团队的后端工程师，他需要同时监听 3 个股票数据源：

"我们的策略需要同时从 NYSE、NASDAQ 和 LSE 三个交易所获取实时报价。之前的 Java 实现用一个线程轮询三个 WebSocket，每次 10ms 的 busy wait，CPU 占用 30%——老板说我比交易系统还费电。"

他打开现在的代码：

GO 
// 坏代码：忙轮询
func pollDataSources() {
    for {
        // 每隔 10ms 轮询一次，浪费 CPU
        if data1 := pollNYSE(); data1 != nil {
            process(data1)
        }
        if data2 := pollNASDAQ(); data2 != nil {
            process(data2)
        }
        if data3 := pollLSE(); data3 != nil {
            process(data3)
        }
        time.Sleep(10 * time.Millisecond)
    }
}

三个问题：(1) 10ms 轮询间隔浪费 CPU；(2) 数据到达和处理不同步；(3) 轮询延迟 0~10ms 不可预测。

(2) Go 的解法：select 监听三个 channel

GO 
// market_data.go
package main

import (
    "fmt"
    "math/rand"
    "time"
)

type Quote struct {
    Source string
    Symbol string
    Price  float64
}

func simulateExchange(name string, out chan<- Quote) {
    symbols := []string{"AAPL", "GOOGL", "MSFT", "AMZN"}
    for {
        time.Sleep(time.Duration(50+rand.Intn(200)) * time.Millisecond)
        quote := Quote{
            Source: name,
            Symbol: symbols[rand.Intn(len(symbols))],
            Price:  100 + rand.Float64()*200,
        }
        out <- quote
    }
}

func main() {
    nyse := make(chan Quote)
    nasdaq := make(chan Quote)
    lse := make(chan Quote)

    go simulateExchange("NYSE", nyse)
    go simulateExchange("NASDAQ", nasdaq)
    go simulateExchange("LSE", lse)

    // select 同时监听三个 channel
    timeout := time.After(2 * time.Second)
    for {
        select {
        case q := <-nyse:
            fmt.Printf("[NYSE] %s: $%.2f\n", q.Symbol, q.Price)
        case q := <-nasdaq:
            fmt.Printf("[NASDAQ] %s: $%.2f\n", q.Symbol, q.Price)
        case q := <-lse:
            fmt.Printf("[LSE] %s: $%.2f\n", q.Symbol, q.Price)
        case <-timeout:
            fmt.Println("演示结束")
            return
        }
    }
}

输出：

TEXT 📖 仅展示 
[NASDAQ] AMZN: $198.32
[NYSE] AAPL: $150.45
[LSE] MSFT: $287.10
...

(3) 收益：select 监听 vs 轮询

方式 CPU 占用 响应延迟 代码复杂度
轮询 10ms 30% 0~10ms 低
轮询 100ms 3% 0~100ms 低
select 监听 0% 0ms（实时） 中
💡 提示： select + channel 是事件驱动——没有数据时 goroutine 休眠（不占 CPU），数据到达后被 runtime 唤醒。这是 Go 并发模型的高效之处。

3. select 基础

(1) select syntax

GO 
select {
case v := <-ch1:
    // ch1 就绪
case v := <-ch2:
    // ch2 就绪
case ch3 <- value:
    // 可以向 ch3 发送（ch3 有空间或有接收方）
default:
    // 所有 channel 都未就绪
}

▶ 示例：select 随机选择

GO 
package main

import (
    "fmt"
)

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

    ch1 <- "来自 ch1"
    ch2 <- "来自 ch2"

    // 两个 channel 同时就绪，select 随机选一个
    for i := 0; i < 2; i++ {
        select {
        case msg := <-ch1:
            fmt.Println(msg)
        case msg := <-ch2:
            fmt.Println(msg)
        }
    }
}
▶ 试一试

输出（每次可能不同）：

TEXT 📖 仅展示 
来自 ch1
来自 ch2

4. time.After 超时控制

(1) 超时模式

GO 
package main

import (
    "fmt"
    "time"
)

func slowOperation() string {
    time.Sleep(2 * time.Second)
    return "结果"
}

func main() {
    ch := make(chan string)
    go func() {
        ch <- slowOperation()
    }()

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

输出：

TEXT 📖 仅展示 
超时！操作超过 1 秒

▶ 示例：逐级超时（分级等待）

GO 📖 仅展示 
package main

import (
    "fmt"
    "time"
)

func fetchFromCache() string {
    time.Sleep(50 * time.Millisecond)
    return "cache-data"
}

func fetchFromDB() string {
    time.Sleep(200 * time.Millisecond)
    return "db-data"
}

func fetchFromAPI() string {
    time.Sleep(500 * time.Millisecond)
    return "api-data"
}

func main() {
    cache := make(chan string, 1)
    db := make(chan string, 1)
    api := make(chan string, 1)

    go func() { cache <- fetchFromCache() }()
    go func() { db <- fetchFromDB() }()
    go func() { api <- fetchFromAPI() }()

    select {
    case r := <-cache:
        fmt.Println("缓存命中:", r)
    case <-time.After(100 * time.Millisecond):
        select {
        case r := <-db:
            fmt.Println("DB 返回:", r)
        case <-time.After(300 * time.Millisecond):
            select {
            case r := <-api:
                fmt.Println("API 返回:", r)
            case <-time.After(600 * time.Millisecond):
                fmt.Println("所有数据源超时！")
            }
        }
    }
}
逻辑代码 41 行（超过 40 行限制，仅展示）
💡 提示： time.After(d) 返回 <-chan time.Time，在 d 时间后发送当前时间。每次 select 调用 time.After 都会创建一个新 timer——如果在循环中频繁调用，记得用 time.NewTimer 并 Stop，避免资源泄漏。

5. default 非阻塞操作

(1) 非阻塞发送/接收

GO 
package main

import (
    "fmt"
)

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

    // 非阻塞接收
    select {
    case v := <-ch:
        fmt.Println("收到:", v)
    default:
        fmt.Println("无数据（非阻塞）")
    }

    // 非阻塞发送
    ch <- 1
    select {
    case ch <- 2:
        fmt.Println("发送成功")
    default:
        fmt.Println("缓冲已满（非阻塞）")
    }
}

输出：

TEXT 📖 仅展示 
无数据（非阻塞）
缓冲已满（非阻塞）

▶ 示例：非阻塞 channel + 轮询熔断

GO 
package main

import (
    "fmt"
    "time"
)

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

    go func() {
        for i := 1; i <= 10; i++ {
            select {
            case ch <- i:
                // 发送成功
            default:
                fmt.Printf("缓冲满，丢弃 %d\n", i)
            }
            time.Sleep(10 * time.Millisecond)
        }
        close(ch)
    }()

    // 慢消费者
    for v := range ch {
        fmt.Printf("处理: %d\n", v)
        time.Sleep(50 * time.Millisecond)
    }
}
▶ 试一试

输出：

TEXT 📖 仅展示 
处理: 1
处理: 2
处理: 3
缓冲满，丢弃 4
缓冲满，丢弃 5
处理: 6
...
🔥 易错： default 会让 select 立即返回——如果所有 channel 都未就绪就执行 default。这非常适合做非阻塞的 channel 操作，但要小心在 for 循环中用 default 会造成 busy loop。

6. for-select 循环与 done channel

(1) done channel 退出模式

GO 
package main

import (
    "fmt"
    "time"
)

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

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

    time.Sleep(1 * time.Second)
    close(done)
    time.Sleep(100 * time.Millisecond)
    fmt.Println("main 退出")
}

▶ 示例：for-select 三种退出方式

GO 
package main

import (
    "fmt"
    "time"
)

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

    // 生产者
    go func() {
        for i := 1; i <= 5; i++ {
            ch <- i
            time.Sleep(100 * time.Millisecond)
        }
        close(ch)
    }()

    // 消费者（for-select 循环）
    go func() {
        for {
            select {
            case v, ok := <-ch:
                if !ok {
                    fmt.Println("方式1: channel 关闭退出")
                    close(done)
                    return
                }
                fmt.Printf("处理: %d\n", v)
            case <-time.After(1 * time.Second):
                fmt.Println("方式2: 超时退出")
                close(done)
                return
            }
        }
    }()

    <-done
    fmt.Println("main 退出")
}
▶ 试一试

(2) for-select 退出模式对比

模式 触发条件 优点 缺点
close(ch) 发送方关闭 channel 自然结束 只能用于接收方
done channel close(done) 信号 灵活，可由外部触发 需要额外 channel
timeout time.After 超时 防卡死 硬超时不够灵活
context ctx.Done() 可传递取消信号 第 17 课深入

7. fan-out / fan-in 模式

(1) fan-out：任务分发

GO 
package main

import (
    "fmt"
    "sync"
)

func fanOut(jobs <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    var wg sync.WaitGroup

    for w := 0; w < workers; w++ {
        wg.Add(1)
        ch := make(chan int, 10)
        channels[w] = ch

        go func(id int, out chan<- int) {
            defer wg.Done()
            defer close(out)
            for job := range jobs {
                result := job * job
                fmt.Printf("Worker %d: %d^2 = %d\n", id, job, result)
                out <- result
            }
        }(w, ch)
    }

    return channels
}

▶ 示例：fan-in：结果聚合

GO 
package main

import (
    "fmt"
    "sync"
)

func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

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

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

    return out
}

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

    // fan-out：分发给 3 个 worker
    workers := fanOut(jobs, 3)

    // fan-in：聚合所有 worker 的结果
    results := fanIn(workers...)

    // 收集结果
    sum := 0
    count := 0
    for r := range results {
        sum += r
        count++
    }
    fmt.Printf("共 %d 个结果, 总和 = %d\n", count, sum)
}
▶ 试一试

输出：

TEXT 📖 仅展示 
Worker 2: 1^2 = 1
Worker 0: 3^2 = 9
Worker 1: 2^2 = 4
Worker 1: 5^2 = 25
Worker 0: 4^2 = 16
Worker 2: 6^2 = 36
共 6 个结果, 总和 = 91

(2) fan-out vs fan-in

模式 方向 用途
fan-out 1 个 channel → N 个 channel 任务分发、并行计算
fan-in N 个 channel → 1 个 channel 结果聚合、日志收集
100% 
flowchart LR
    subgraph FanOut [Fan-Out]
        J[Jobs Channel] --> W1[Worker 1]
        J --> W2[Worker 2]
        J --> W3[Worker 3]
    end
    subgraph FanIn [Fan-In]
        W1 --> R[Results Channel]
        W2 --> R
        W3 --> R
    end
    style J fill:#e1f5fe
    style R fill:#fff3e0

8. 完整示例：三数据源股票行情聚合

GO 
// market_aggregator.go
package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// ---------- 数据类型 ----------

type Trade struct {
    Source   string
    Symbol   string
    Price    float64
    Volume   int
    Time     time.Time
}

type AggregatedQuote struct {
    Symbol      string
    AvgPrice    float64
    TotalVolume int
    Sources     int
    High        float64
    Low         float64
    Time        time.Time
}

// ---------- 模拟交易所数据源 ----------

var symbols = []string{"AAPL", "GOOGL", "MSFT", "AMZN", "TSLA", "META"}

func simulateExchange(name string, out chan<- Trade, done <-chan struct{}) {
    for {
        select {
        case <-done:
            return
        default:
            time.Sleep(time.Duration(100+rand.Intn(300)) * time.Millisecond)
            trade := Trade{
                Source: name,
                Symbol: symbols[rand.Intn(len(symbols))],
                Price:  100 + rand.Float64()*200,
                Volume: rand.Intn(1000) + 100,
                Time:   time.Now(),
            }
            select {
            case out <- trade:
            case <-done:
                return
            }
        }
    }
}

// ---------- 聚合器 ----------

type Aggregator struct {
    trades chan Trade
    done   chan struct{}
    wg     sync.WaitGroup
}

func NewAggregator() *Aggregator {
    return &Aggregator{
        trades: make(chan Trade, 100),
        done:   make(chan struct{}),
    }
}

func (a *Aggregator) AddSource(name string) {
    a.wg.Add(1)
    go func() {
        defer a.wg.Done()
        simulateExchange(name, a.trades, a.done)
    }()
    fmt.Printf("添加数据源: %s\n", name)
}

func (a *Aggregator) Start(interval time.Duration, callback func(map[string]*AggregatedQuote)) {
    // 聚合窗口
    window := make(map[string][]Trade)

    ticker := time.NewTicker(interval)
    defer ticker.Stop()

    for {
        select {
        case trade := <-a.trades:
            window[trade.Symbol] = append(window[trade.Symbol], trade)

        case <-ticker.C:
            // 时间窗口到，计算聚合报价
            quotes := make(map[string]*AggregatedQuote)

            for symbol, trades := range window {
                if len(trades) == 0 {
                    continue
                }
                var sumPrice float64
                var totalVol int
                high := trades[0].Price
                low := trades[0].Price

                for _, t := range trades {
                    sumPrice += t.Price * float64(t.Volume)
                    totalVol += t.Volume
                    if t.Price > high {
                        high = t.Price
                    }
                    if t.Price < low {
                        low = t.Price
                    }
                }

                sources := make(map[string]bool)
                for _, t := range trades {
                    sources[t.Source] = true
                }

                quotes[symbol] = &AggregatedQuote{
                    Symbol:      symbol,
                    AvgPrice:    sumPrice / float64(totalVol),
                    TotalVolume: totalVol,
                    Sources:     len(sources),
                    High:        high,
                    Low:         low,
                    Time:        time.Now(),
                }
            }

            callback(quotes)
            window = make(map[string][]Trade) // 重置窗口

        case <-a.done:
            return
        }
    }
}

func (a *Aggregator) Stop() {
    close(a.done)
    a.wg.Wait()
}

// ---------- 主函数 ----------

func main() {
    aggregator := NewAggregator()

    // 添加 3 个数据源
    aggregator.AddSource("NYSE")
    aggregator.AddSource("NASDAQ")
    aggregator.AddSource("LSE")

    fmt.Println("\n开始聚合（每 2 秒输出一次）...\n")

    // 启动聚合，2 秒时间窗口
    done := make(chan struct{})
    go func() {
        aggregator.Start(2*time.Second, func(quotes map[string]*AggregatedQuote) {
            fmt.Printf("=== 聚合报告 %s ===\n", time.Now().Format("15:04:05"))
            for _, q := range quotes {
                fmt.Printf("%-6s | $%.2f (avg) | vol=%d | src=%d | H=%.2f L=%.2f\n",
                    q.Symbol, q.AvgPrice, q.TotalVolume, q.Sources, q.High, q.Low)
            }
            fmt.Println()
        })
        close(done)
    }()

    // 运行 10 秒后停止
    time.Sleep(10 * time.Second)
    aggregator.Stop()
    <-done
    fmt.Println("行情聚合器已停止")
}

预期输出：

TEXT 📖 仅展示 
添加数据源: NYSE
添加数据源: NASDAQ
添加数据源: LSE

开始聚合（每 2 秒输出一次）...

=== 聚合报告 10:00:02 ===
AAPL   | $152.34 (avg) | vol=2341 | src=3 | H=165.20 L=142.10
GOOGL  | $178.90 (avg) | vol=1567 | src=2 | H=185.00 L=172.30
MSFT   | $295.40 (avg) | vol=890  | src=3 | H=301.20 L=288.50

=== 聚合报告 10:00:04 ===
...

行情聚合器已停止
🔥 易错： select 中 case <-done: 和 case trade := <-a.trades: 需要同时监听。如果只监听 trades channel，Stop 信号就无法送达。任何 for-select 循环都必须包含退出条件。

❓ 常见问题

Q select 多个 channel 同时就绪怎么办？
A Go 会伪随机选择一个执行的 case。这是语言规范——防止开发者依赖 channel 优先级。如果你需要优先级，用两个嵌套 select 或 default + 手动检查。
Q time.After 如何做超时？
A time.After(d) 返回一个 channel，在 d 时间后收到值。在 select 中用 case <-time.After(d): 作为超时分支。注意：每次 select 调用 time.After 都会创建新 timer，循环中建议用 time.NewTimer 复用。
Q default 在 select 中起什么作用？
A default 让 select 变成非阻塞——所有 channel 都未就绪时立即执行 default。适合：尝试发送/接收但不阻塞、统计失败率、熔断降级。注意：for 循环中有 default 会变成 busy loop，通常需要加 time.Sleep 或限制频率。
Q 如何用 done channel 退出 for-select 循环？
A 创建 done := make(chan struct{})，select 中监听 case <-done:，需要退出时 close(done)。所有监听 done 的 goroutine 会同时收到退出信号。close 比发送一个值更优——可被多次接收。
Q fan-in 和 fan-out 是什么？
A fan-out（扇出）= 1 个输入 channel 分发到 N 个 worker goroutine（并行处理）；fan-in（扇入）= N 个 channel 聚合到 1 个输出 channel（合并结果）。两者组合是 Go 并行计算的经典模式。
Q select 可以用于发送吗？
A 可以。case ch <- value: 也是合法的 select case。当 channel 有空间（有缓冲）或有接收方（无缓冲）时就绪。这个特性用于限流和信号量模式。
Q 空的 select{} 会怎样？
A select {} 会永久阻塞——因为没有任何 case，且没有 default。这可以作为"永久等待"的信号。但生产代码中应该确保有退出路径，否则 goroutine 泄漏。
Q for-select 中 break 能跳出循环吗？
A 不能。select 中的 break 只会跳出 select 块，不会跳出 for 循环。要跳出 for-select，用 label + break label，或 return，或 done channel。

📖 小节

📝 作业

  1. 基础题（难度⭐）：写一个程序启动 3 个 goroutine，分别发送整数到各自的 channel，用 select 同时接收并打印。添加一个超时分支，3 秒后退出。

  2. 进阶题（难度⭐⭐）：实现一个超时缓存：func FetchWithCache(key string, cache map[string]string) string，先查缓存 channel（10ms 超时），未命中则从"模拟数据库"（500ms）查询并回填缓存。要求用 select + time.After 实现分级超时。

  3. 挑战题（难度⭐⭐⭐）：实现一个日志聚合系统：3 个日志源（goroutine）各自生成不同级别的日志（INFO/WARN/ERROR），用 fan-in 聚合到一个 channel，然后用 select 做分级处理——ERROR 立即告警（打印），WARN 转存（5 条一刷），INFO 批量写入（10 条一刷）。要求：超时 flush + done channel 优雅退出。

Web-Tutorial.com

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

100%

🙏 帮我们做得更好

我们是刚上线的编程教程站，几个人的小团队，精力有限。页面虽经检查，难免还有疏漏——链接失效、排版错乱、内容有误、语言生硬……

如果您发现了，麻烦告诉我们，我们会在收到反馈后第一时间进行修复，再次感谢您的光临 🙏