Goroutine、Channel、Select——Go 并发的核心原语

一、Goroutine

1.1 什么是 Goroutine

Goroutine 是 Go 运行时管理的轻量级线程。创建成本极低（约 2KB 栈空间），可以轻松创建成千上万个。

// 启动一个 goroutine：在函数调用前加 go 关键字
go doSomething()
go func() {
    fmt.Println("running in goroutine")
}()

1.2 基本使用

package main

import (
    "fmt"
    "time"
)

func printNumbers() {
    for i := 1; i <= 5; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("number: %d\n", i)
    }
}

func printLetters() {
    for i := 'a'; i <= 'e'; i++ {
        time.Sleep(150 * time.Millisecond)
        fmt.Printf("letter: %c\n", i)
    }
}

func main() {
    go printNumbers()   // 并发执行
    go printLetters()   // 并发执行

    time.Sleep(2 * time.Second)  // 等待 goroutine 完成
    fmt.Println("done")
}
// 输出交错出现：number: 1, letter: a, number: 2, letter: b...

1.3 主 Goroutine 与 WaitGroup

上面用 time.Sleep 等待是不可靠的。正确做法是用 sync.WaitGroup：

import "sync"

func main() {
    var wg sync.WaitGroup

    wg.Add(2)  // 声明要等待的 goroutine 数量

    go func() {
        defer wg.Done()  // 完成时通知 WaitGroup
        printNumbers()
    }()

    go func() {
        defer wg.Done()
        printLetters()
    }()

    wg.Wait()  // 阻塞直到所有 goroutine 完成
    fmt.Println("done")
}

重要：main goroutine 退出时，所有其他 goroutine 会被强制终止。不要让 main 提前退出。

二、Channel

Channel 是 goroutine 之间的通信管道，用于传递数据和同步。

2.1 基本用法

// 创建 channel
ch := make(chan int)     // 无缓冲 channel（同步）
ch := make(chan int, 5)  // 有缓冲 channel（异步，容量5）

// 发送数据
ch <- 42        // 将 42 发送到 channel

// 接收数据
value := <-ch   // 从 channel 接收数据
<-ch            // 丢弃接收到的值

// 关闭 channel
close(ch)       // 发送方关闭，接收方可以继续读取剩余数据
// 只有发送方应该关闭 channel，接收方关闭会 panic

2.2 无缓冲 Channel（同步）

// 无缓冲 channel 是同步的：发送方阻塞直到接收方准备好
func main() {
    ch := make(chan string)

    go func() {
        ch <- "hello from goroutine"  // 发送，阻塞等待接收
    }()

    msg := <-ch  // 接收，阻塞等待发送
    fmt.Println(msg)
}

2.3 有缓冲 Channel（异步）

// 有缓冲 channel：缓冲区满时发送方阻塞，空时接收方阻塞
ch := make(chan int, 3)  // 缓冲区大小为 3

ch <- 1   // 不阻塞
ch <- 2   // 不阻塞
ch <- 3   // 不阻塞
// ch <- 4 // 阻塞！缓冲区已满

fmt.Println(<-ch)  // 1
fmt.Println(<-ch)  // 2
fmt.Println(<-ch)  // 3

2.4 Channel 方向

// 只能发送的 channel
func producer(ch chan<- int) {
    ch <- 42
    close(ch)
}

// 只能接收的 channel
func consumer(ch <-chan int) {
    for v := range ch {
        fmt.Println(v)
    }
}

// 双向 channel 可以隐式转换为单向 channel
ch := make(chan int)
go producer(ch)  // chan int -> chan<- int
go consumer(ch)  // chan int -> <-chan int

2.5 遍历 Channel

// range 遍历 channel，直到 channel 关闭
ch := make(chan int, 5)
ch <- 1
ch <- 2
ch <- 3
close(ch)

for v := range ch {
    fmt.Println(v)  // 1, 2, 3
}
// channel 关闭后 range 自动退出

三、Select 多路复用

select 用于同时监听多个 channel 操作，类似 switch 但针对 channel。

3.1 基本用法

select {
case msg := <-ch1:
    fmt.Println("received from ch1:", msg)
case msg := <-ch2:
    fmt.Println("received from ch2:", msg)
case ch3 <- value:
    fmt.Println("sent to ch3")
default:
    fmt.Println("no channel ready")
}

3.2 超时控制

func fetchData(ch chan string) string {
    select {
    case result := <-ch:
        return result
    case <-time.After(3 * time.Second):
        return "timeout"
    }
}

3.3 非阻塞操作

// default 分支使 select 成为非阻塞
select {
case msg := <-ch:
    fmt.Println("got:", msg)
default:
    fmt.Println("no message available")
}

3.4 done channel 模式

func worker(done chan bool) {
    fmt.Println("working...")
    time.Sleep(time.Second)
    done <- true  // 通知完成
}

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

    select {
    case <-done:
        fmt.Println("worker finished")
    case <-time.After(5 * time.Second):
        fmt.Println("timeout waiting for worker")
    }
}

四、经典并发模式

4.1 Fan-Out / Fan-In

// Fan-Out：多个 goroutine 从同一 channel 读取
// Fan-In：多个 goroutine 写入同一 channel

func fanOut(input <-chan int, workers int) []<-chan int {
    channels := make([]<-chan int, workers)
    for i := 0; i < workers; i++ {
        channels[i] = process(input)
    }
    return channels
}

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

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

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

    return merged
}

4.2 工作池（Worker Pool）

func workerPool(jobs <-chan Job, results chan<- Result, numWorkers int) {
    var wg sync.WaitGroup

    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                result := process(job)
                results <- result
            }
        }(i)
    }

    wg.Wait()
    close(results)
}

4.3 Context 超时与取消

import "context"

func longOperation(ctx context.Context) error {
    select {
    case <-time.After(10 * time.Second):
        fmt.Println("operation completed")
        return nil
    case <-ctx.Done():
        return ctx.Err()  // context canceled 或 deadline exceeded
    }
}

// 使用
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
defer cancel()

err := longOperation(ctx)
if err != nil {
    fmt.Println("operation canceled:", err)
}

五、sync 包常用工具

5.1 Mutex 互斥锁

import "sync"

type SafeCounter struct {
    mu    sync.Mutex
    count int
}

func (c *SafeCounter) Increment() {
    c.mu.Lock()
    defer c.mu.Unlock()
    c.count++
}

func (c *SafeCounter) Get() int {
    c.mu.Lock()
    defer c.mu.Unlock()
    return c.count
}

5.2 RWMutex 读写锁

type Cache struct {
    mu   sync.RWMutex
    data map[string]string
}

func (c *Cache) Get(key string) (string, bool) {
    c.mu.RLock()         // 读锁（多个读可以并发）
    defer c.mu.RUnlock()
    v, ok := c.data[key]
    return v, ok
}

func (c *Cache) Set(key, value string) {
    c.mu.Lock()          // 写锁（独占）
    defer c.mu.Unlock()
    c.data[key] = value
}

5.3 Once — 只执行一次

var once sync.Once
var instance *Database

func GetDB() *Database {
    once.Do(func() {
        instance = connectDB()  // 只会执行一次，即使多个 goroutine 并发调用
    })
    return instance
}

5.4 atomic — 原子操作

import "sync/atomic"

var count int64

// 原子加操作，无需锁
atomic.AddInt64(&count, 1)
atomic.LoadInt64(&count)
atomic.StoreInt64(&count, 0)

六、并发陷阱

6.1 Race Condition（竞态条件）

// 多个 goroutine 同时读写共享变量 = 数据竞争
count := 0
for i := 0; i < 1000; i++ {
    go func() {
        count++  // 数据竞争！
    }()
}

// 正确：使用 atomic 或 channel 或 Mutex
var count int64
for i := 0; i < 1000; i++ {
    go func() {
        atomic.AddInt64(&count, 1)
    }()
}

// 检测竞态条件
// go run -race main.go
// go test -race ./...

6.2 Goroutine 泄漏

// goroutine 永远阻塞，无法退出 = 泄漏
func leak() {
    ch := make(chan int)
    go func() {
        ch <- 42  // 没有人接收，永远阻塞
    }()
}

// 正确：使用 context 或 done channel 确保 goroutine 可以退出
func noLeak(ctx context.Context) {
    ch := make(chan int)
    go func() {
        select {
        case ch <- 42:
        case <-ctx.Done():
            return  // context 取消时退出
        }
    }()
}

6.3 关闭已关闭的 Channel

ch := make(chan int)
close(ch)
// close(ch)  // panic: close of closed channel

// 如果不确定是否已关闭，不要关闭 channel
// 或者只由发送方关闭

七、本节要点速查

概念要点
goroutinego func() 启动，轻量级（~2KB）
WaitGroupAdd/Done/Wait 等待 goroutine 完成
无缓冲 channelmake(chan T) 同步通信
有缓冲 channelmake(chan T, n) 异步通信
channel 方向chan<- 只发，<-chan 只收
select多路复用，可加 default/timeout
Context超时控制与取消传播
MutexLock/Unlock 互斥锁
RWMutex多读单写锁
Once只执行一次（单例模式）
atomic无锁原子操作
go run -race检测竞态条件

上一篇：08-包与模块 下一篇：10-测试与工程实践