Golang协程并发goroutine详解

Posted on Views:

Go 的协程（goroutine）是其并发模型的核心，轻量级、无栈（堆栈通常几 KB），线程级别的分身。创建成本极低（~100–200 字节），可以轻松创建几百万个goroutine而不会耗尽系统资源。

1. goroutine启动与生命周期

1
2
3
go func() {  // 匿名函数启动 goroutine
    // 代码在这里并行执行
}()
  • go 关键字必须在函数/方法调用后面。
  • 启动的 goroutine 是异步的，当前主 goroutine 立即继续执行，不会等待。
  • 主 goroutine 结束时，整个程序退出（即使其他 goroutine 还在跑）。

经典启动方式（推荐）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
func worker(id int) {
    fmt.Printf("worker %d started\n", id)
}

func main() {
    var wg sync.WaitGroup
    for i := 0; i < 5; i++ {
        wg.Add(1)
        go func(id int) {  // 必须捕获 id，否则闭包问题！
            defer wg.Done()
            worker(id)
        }(i)  // 传参
    }
    wg.Wait()
}

2. 常见并发问题与解决方案

问题 原因 推荐解决方案
数据竞争 多个 goroutine 同时读写变量 sync.Mutex、sync.RWMutex、原子操作（sync/atomic）
死锁 通道永久阻塞或锁顺序错误 select + default 超时、context 取消
goroutine 泄漏 忘记 wg.Done() 或取消信号 sync.WaitGroup + context
主线程退出导致程序退出 其他 goroutine 没结束 必须显式等待（wg.Wait()、context）

3. 经典并发模式

模式1：流水线（Pipeline）—— 最推荐的 CSP 风格

上游产生数据 → 中间 goroutine 加工 → 下游消费。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
func pipeline() {
    jobs := make(chan int, 10)
    results := make(chan int, 10)

    var wg sync.WaitGroup
    wg.Add(2)

    go func() {
        defer wg.Done()
        for i := 1; i <= 5; i++ {
            jobs <- i
        }
        close(jobs)  // 关闭信号向下游传递
    }()

    go func() {
        defer wg.Done()
        for j := range results {
            fmt.Println("processed:", j)
        }
    }()

    wg.Wait()
    close(results)
}

模式 2：Worker Pool（工人池）—— 控制并发数量

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
func workerPool() {
    const numWorkers = 10
    jobs := make(chan int, 100)
    results := make(chan string, 100)

    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for job := range jobs {
                results <- fmt.Sprintf("worker %d processed job %d", id, job)
            }
        }(i)
    }

    for j := 1; j <= 20; j++ {
        jobs <- j
    }
    close(jobs)

    wg.Wait()
    close(results)

    for r := range results {
        fmt.Println(r)
    }
}

模式 3：取消信号（Cancel）—— 优雅退出

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
func workerWithCancel(ctx context.Context, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
    defer wg.Done()
    for {
        select {
        case v, ok := <-jobs:
            if !ok {
                return
            }
            results <- v * 2
        case <-ctx.Done():
            return
        }
    }
}

func main() {
    ctx, cancel := context.WithCancel(context.Background())
    var wg sync.WaitGroup
    jobs := make(chan int, 100)
    results := make(chan int, 100)

    for i := 1; i <= 5; i++ {
        wg.Add(1)
        go workerWithCancel(ctx, jobs, results, &wg)
    }

    for j := 1; j <= 10; j++ {
        jobs <- j
    }
    close(jobs)

    // 10ms 后取消
    time.AfterFunc(10*time.Millisecond, cancel)

    wg.Wait()
    close(results)
    for r := range results {
        fmt.Println(r)
    }
}

4. goroutine底层原理

  • 每个 goroutine 有一个栈（通常几 KB），支持动态扩展。
  • Go 运行时使用抢占式调度（runtime scheduler），支持抢占式协程（preemptive goroutines）。
  • 调度器使用work-stealing 和 global runqueue。
  • runtime.Gosched() 可以手动让出 CPU（很少用）。
  • 通道底层是 chan 结构体 + 运行时锁，操作都是原子化的。