1
并发模式分类
2
│
3
├── 数据流模式
4
│   ├── Pipeline（流水线）
5
│   ├── Fan-out/Fan-in（扇出/扇入）
6
│   └── Scatter-Gather（分发-聚合）
7
│
8
├── 资源管理模式
9
│   ├── Worker Pool（工作池）
10
│   ├── Semaphore（信号量）
11
│   └── Bounded Parallelism（有界并行）
12
│
13
├── 控制模式
14
│   ├── Context Cancellation（上下文取消）
15
│   ├── Timeout（超时控制）
16
│   └── Graceful Shutdown（优雅关闭）
17
│
18
└── 错误处理模式
19
    ├── errgroup（错误组）
20
    ├── First Error Wins（首个错误胜出）
21
    └── Error Aggregation（错误聚合）

1
// Stage 1: 生成器——产生数据源
2
func generate(nums ...int) <-chan int {
3
    out := make(chan int)
4
    go func() {
5
        defer close(out)
6
        for _, n := range nums {
7
            out <- n
8
        }
9
    }()
10
    return out
11
}
12

13
// Stage 2: 平方运算——转换数据
14
func square(in <-chan int) <-chan int {
15
    out := make(chan int)
16
    go func() {
17
        defer close(out)
18
        for n := range in {
19
            out <- n * n
20
        }
21
    }()
22
    return out
23
}
24

25
// Stage 3: 过滤——筛选符合条件的值
26
func filter(in <-chan int, predicate func(int) bool) <-chan int {
27
    out := make(chan int)
28
    go func() {
29
        defer close(out)
30
        for n := range in {
31
            if predicate(n) {
32
                out <- n
33
            }
34
        }
35
    }()
36
    return out
37
}
38

39
// 使用示例
40
func main() {
41
    // 构建流水线
42
    nums := generate(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
43
    squares := square(nums)
44
    evens := filter(squares, func(n int) bool { return n > 25 })
45

46
    // 消费结果
47
    for n := range evens {
48
        fmt.Println(n)
49
    }
50
}

1
func generateContext(ctx context.Context, nums ...int) <-chan int {
2
    out := make(chan int)
3
    go func() {
4
        defer close(out)
5
        for _, n := range nums {
6
            select {
7
            case out <- n:
8
            case <-ctx.Done():
9
                return // 取消时退出
10
            }
11
        }
12
    }()
13
    return out
14
}
15

16
func squareContext(ctx context.Context, in <-chan int) <-chan int {
17
    out := make(chan int)
18
    go func() {
19
        defer close(out)
20
        for {
21
            select {
22
            case n, ok := <-in:
23
                if !ok {
24
                    return
25
                }
26
                select {
27
                case out <- n * n:
28
                case <-ctx.Done():
29
                    return
30
                }
31
            case <-ctx.Done():
32
                return
33
            }
34
        }
35
    }()
36
    return out
37
}

1
// 并行执行多个 worker
2
func fanOut(worker func(<-chan Task) <-chan Result, jobs <-chan Task, workers int) []<-chan Result {
3
    results := make([]<-chan Result, workers)
4
    for i := 0; i < workers; i++ {
5
        results[i] = worker(jobs)
6
    }
7
    return results
8
}
9

10
func worker(jobs <-chan Task) <-chan Result {
11
    out := make(chan Result)
12
    go func() {
13
        defer close(out)
14
        for job := range jobs {
15
            out <- process(job)
16
        }
17
    }()
18
    return out
19
}

1
// 合并多个 channel 的结果
2
func fanIn(ctx context.Context, channels ...<-chan Result) <-chan Result {
3
    out := make(chan Result)
4

5
    var wg sync.WaitGroup
6
    wg.Add(len(channels))
7

8
    // 为每个输入 channel 启动一个 goroutine
9
    for _, ch := range channels {
10
        go func(c <-chan Result) {
11
            defer wg.Done()
12
            for result := range c {
13
                select {
14
                case out <- result:
15
                case <-ctx.Done():
16
                    return
17
                }
18
            }
19
        }(ch)
20
    }
21

22
    // 等待所有 channel 关闭后关闭输出
23
    go func() {
24
        wg.Wait()
25
        close(out)
26
    }()
27

28
    return out
29
}

1
func fetchURLs(urls []string) ([]Page, error) {
2
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
3
    defer cancel()
4

5
    // 创建任务 channel
6
    jobs := make(chan string, len(urls))
7
    for _, url := range urls {
8
        jobs <- url
9
    }
10
    close(jobs)
11

12
    // Fan-out：启动多个 worker 并行抓取
13
    numWorkers := min(5, len(urls))
14
    results := fanOut(fetchWorker, jobs, numWorkers)
15

16
    // Fan-in：合并结果
17
    merged := fanIn(ctx, results...)
18

19
    // 收集结果
20
    var pages []Page
21
    for result := range merged {
22
        if result.Err != nil {
23
            return nil, result.Err
24
        }
25
        pages = append(pages, result.Page)
26
    }
27

28
    return pages, nil
29
}
30

31
func fetchWorker(jobs <-chan string) <-chan FetchResult {
32
    out := make(chan FetchResult)
33
    go func() {
34
        defer close(out)
35
        client := http.Client{Timeout: 10 * time.Second}
36
        for url := range jobs {
37
            resp, err := client.Get(url)
38
            if err != nil {
39
                out <- FetchResult{Err: err}
40
                continue
41
            }
42
            body, _ := io.ReadAll(resp.Body)
43
            resp.Body.Close()
44
            out <- FetchResult{Page: Page{URL: url, Content: string(body)}}
45
        }
46
    }()
47
    return out
48
}

1
type WorkerPool struct {
2
    tasks   chan Task
3
    results chan Result
4
    workers int
5
    wg      sync.WaitGroup
6
}
7

8
func NewWorkerPool(workers, taskCapacity int) *WorkerPool {
9
    return &WorkerPool{
10
        tasks:   make(chan Task, taskCapacity),
11
        results: make(chan Result, taskCapacity),
12
        workers: workers,
13
    }
14
}
15

16
func (p *WorkerPool) Start() {
17
    for i := 0; i < p.workers; i++ {
18
        p.wg.Add(1)
19
        go p.worker(i)
20
    }
21
}
22

23
func (p *WorkerPool) worker(id int) {
24
    defer p.wg.Done()
25
    for task := range p.tasks {
26
        result := process(task)
27
        p.results <- result
28
    }
29
}
30

31
func (p *WorkerPool) Submit(task Task) {
32
    p.tasks <- task
33
}
34

35
func (p *WorkerPool) Stop() {
36
    close(p.tasks)
37
    p.wg.Wait()
38
    close(p.results)
39
}
40

41
func (p *WorkerPool) Results() <-chan Result {
42
    return p.results
43
}

1
type CancellableWorkerPool struct {
2
    tasks   chan Task
3
    results chan Result
4
    workers int
5
    ctx     context.Context
6
    cancel  context.CancelFunc
7
    wg      sync.WaitGroup
8
}
9

10
func NewCancellableWorkerPool(ctx context.Context, workers, capacity int) *CancellableWorkerPool {
11
    childCtx, cancel := context.WithCancel(ctx)
12
    return &CancellableWorkerPool{
13
        tasks:   make(chan Task, capacity),
14
        results: make(chan Result, capacity),
15
        workers: workers,
16
        ctx:     childCtx,
17
        cancel:  cancel,
18
    }
19
}
20

21
func (p *CancellableWorkerPool) Start() {
22
    for i := 0; i < p.workers; i++ {
23
        p.wg.Add(1)
24
        go p.worker(i)
25
    }
26
}
27

28
func (p *CancellableWorkerPool) worker(id int) {
29
    defer p.wg.Done()
30
    for {
31
        select {
32
        case task, ok := <-p.tasks:
33
            if !ok {
34
                return
35
            }
36
            result := p.processWithCancel(task)
37
            select {
38
            case p.results <- result:
39
            case <-p.ctx.Done():
40
                return
41
            }
42
        case <-p.ctx.Done():
43
            return
44
        }
45
    }
46
}
47

48
func (p *CancellableWorkerPool) processWithCancel(task Task) Result {
49
    // 在处理过程中也检查取消信号
50
    done := make(chan Result, 1)
51
    go func() {
52
        done <- process(task)
53
    }()
54

55
    select {
56
    case result := <-done:
57
        return result
58
    case <-p.ctx.Done():
59
        return Result{Err: p.ctx.Err()}
60
    }
61
}
62

63
func (p *CancellableWorkerPool) Cancel() {
64
    p.cancel()
65
}
66

67
func (p *CancellableWorkerPool) Stop() {
68
    close(p.tasks)
69
    p.wg.Wait()
70
    close(p.results)
71
}

1
// 危险：无限制创建 goroutine
2
func dangerous(urls []string) {
3
    for _, url := range urls {
4
        go fetch(url) // 可能创建数千个 goroutine
5
    }
6
}
7

8
// 安全：使用 Worker Pool 限制并发
9
func safe(urls []string, maxWorkers int) {
10
    pool := NewWorkerPool(maxWorkers, len(urls))
11
    pool.Start()
12

13
    for _, url := range urls {
14
        pool.Submit(Task{URL: url})
15
    }
16

17
    go func() {
18
        pool.Stop()
19
    }()
20

21
    for result := range pool.Results() {
22
        // 处理结果
23
    }
24
}

1
type Context interface {
2
    // 返回 context 被取消的截止时间
3
    Deadline() (deadline time.Time, ok bool)
4

5
    // 返回一个 channel，当 context 被取消时关闭
6
    Done() <-chan struct{}
7

8
    // 返回取消原因
9
    Err() error
10

11
    // 获取与 context 关联的值
12
    Value(key any) any
13
}

1
func operation(ctx context.Context) error {
2
    // 启动子任务
3
    ctx, cancel := context.WithCancel(ctx)
4
    defer cancel() // 确保资源释放
5

6
    // 启动多个子任务
7
    results := make(chan Result, 3)
8
    for i := 0; i < 3; i++ {
9
        go func(id int) {
10
            results <- subOperation(ctx, id)
11
        }(i)
12
    }
13

14
    // 等待第一个结果或取消
15
    select {
16
    case result := <-results:
17
        cancel() // 取消其他子任务
18
        return result.Err
19
    case <-ctx.Done():
20
        return ctx.Err()
21
    }
22
}
23

24
func subOperation(ctx context.Context, id int) Result {
25
    // 定期检查取消信号
26
    for i := 0; i < 100; i++ {
27
        select {
28
        case <-ctx.Done():
29
            return Result{Err: ctx.Err()}
30
        default:
31
            // 执行工作
32
            time.Sleep(10 * time.Millisecond)
33
        }
34
    }
35
    return Result{Value: id}
36
}

1
// 正确：将 Context 作为第一个参数
2
func DoSomething(ctx context.Context, arg Arg) error {
3
    // ...
4
}
5

6
// 错误：将 Context 放在其他位置
7
func DoSomething(arg Arg, ctx context.Context) error {
8
    // ...
9
}
10

11
// 正确：不要将 Context 存储在结构体中（少数例外：HTTP Handler）
12
type Handler struct {
13
    // ctx 不应该作为字段
14
}
15

16
// 正确：context.Background() 作为根 Context
17
ctx := context.Background()
18

19
// 正确：传递取消原因
20
func process(ctx context.Context) error {
21
    ctx, cancel := context.WithCancelCause(ctx)
22
    go func() {
23
        if err := checkSomething(); err != nil {
24
            cancel(fmt.Errorf("check failed: %w", err))
25
        }
26
    }()
27
    <-ctx.Done()
28
    return context.Cause(ctx) // 获取具体的取消原因
29
}

1
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
2
    // 创建带超时的 context
3
    ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
4
    defer cancel()
5

6
    req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
7
    if err != nil {
8
        return nil, err
9
    }
10

11
    resp, err := http.DefaultClient.Do(req)
12
    if err != nil {
13
        if errors.Is(err, context.DeadlineExceeded) {
14
            return nil, fmt.Errorf("request timed out: %w", err)
15
        }
16
        return nil, err
17
    }
18

19
    return resp, nil
20
}

1
func simpleTimeout(timeout time.Duration) error {
2
    result := make(chan error, 1)
3

4
    go func() {
5
        result <- doWork()
6
    }()
7

8
    select {
9
    case err := <-result:
10
        return err
11
    case <-time.After(timeout):
12
        return fmt.Errorf("operation timed out after %v", timeout)
13
    }
14
}

1
超时控制流程：
2

3
开始 ────▶ 执行任务 ────▶ 完成
4
   │
5
   │  (超过指定时间)
6
   │
7
   └───────▶ 返回 DeadlineExceeded 错误
8

9

10
取消控制流程：
11

12
开始 ────▶ 执行任务 ────▶ 完成
13
   │
14
   │  (调用 cancel())
15
   │
16
   └───────▶ 返回 Canceled 错误

1
func multiLayerTimeout() error {
2
    // 外层：总超时 30 秒
3
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
4
    defer cancel()
5

6
    // 阶段1：数据库查询，超时 10 秒
7
    dbCtx, dbCancel := context.WithTimeout(ctx, 10*time.Second)
8
    data, err := queryDB(dbCtx)
9
    dbCancel()
10
    if err != nil {
11
        return fmt.Errorf("database query: %w", err)
12
    }
13

14
    // 阶段2：外部 API 调用，超时 15 秒
15
    apiCtx, apiCancel := context.WithTimeout(ctx, 15*time.Second)
16
    result, err := callAPI(apiCtx, data)
17
    apiCancel()
18
    if err != nil {
19
        return fmt.Errorf("api call: %w", err)
20
    }
21

22
    return nil
23
}

1
import "golang.org/x/sync/errgroup"
2

3
func fetchAll(urls []string) ([]Page, error) {
4
    g, ctx := errgroup.WithContext(context.Background())
5
    pages := make([]Page, len(urls))
6

7
    for i, url := range urls {
8
        i, url := i, url // 捕获循环变量
9
        g.Go(func() error {
10
            // 如果其他 goroutine 出错，ctx 会被取消
11
            page, err := fetchPage(ctx, url)
12
            if err != nil {
13
                return err
14
            }
15
            pages[i] = page
16
            return nil
17
        })
18
    }
19

20
    // 等待所有任务完成，返回第一个错误（如果有）
21
    if err := g.Wait(); err != nil {
22
        return nil, err
23
    }
24

25
    return pages, nil
26
}

1
errgroup 执行流程：
2

3
┌─────────────────────────────────────────────────────┐
4
│                     errgroup                         │
5
│                                                     │
6
│   ┌─────────┐   ┌─────────┐   ┌─────────┐          │
7
│   │ Task 1  │   │ Task 2  │   │ Task 3  │          │
8
│   │ (成功)  │   │ (出错)  │   │ (运行中)│          │
9
│   └────┬────┘   └────┬────┘   └────┬────┘          │
10
│        │             │             │               │
11
│        │         ┌───▼───┐         │               │
12
│        │         │ error │◀────────┘               │
13
│        │         └───┬───┘  ctx 被取消              │
14
│        │             │                             │
15
│        └─────────────┼─────────────────┐           │
16
│                      ▼                 │           │
17
│               ┌─────────────┐          │           │
18
│               │ g.Wait()    │◀─────────┘           │
19
│               │ 返回 error  │                      │
20
│               └─────────────┘                      │
21
└─────────────────────────────────────────────────────┘

1
func fetchWithLimit(urls []string, maxConcurrent int) ([]Page, error) {
2
    g, ctx := errgroup.WithContext(context.Background())
3

4
    // 使用信号量限制并发数
5
    sem := make(chan struct{}, maxConcurrent)
6
    pages := make([]Page, len(urls))
7

8
    for i, url := range urls {
9
        i, url := i, url
10
        sem <- struct{}{} // 获取信号量
11

12
        g.Go(func() error {
13
            defer func() { <-sem }() // 释放信号量
14

15
            page, err := fetchPage(ctx, url)
16
            if err != nil {
17
                return err
18
            }
19
            pages[i] = page
20
            return nil
21
        })
22
    }
23

24
    if err := g.Wait(); err != nil {
25
        return nil, err
26
    }
27

28
    return pages, nil
29
}

1
func monitorGoroutines() {
2
    ticker := time.NewTicker(5 * time.Second)
3
    for range ticker.C {
4
        count := runtime.NumGoroutine()
5
        log.Printf("current goroutines: %d", count)
6
        if count > 100 {
7
            log.Printf("WARNING: too many goroutines!")
8
        }
9
    }
10
}

1
// 场景1：无缓冲 channel 阻塞
2
func leak1() {
3
    ch := make(chan int)
4
    go func() {
5
        ch <- 42 // 没有接收者，永久阻塞
6
    }()
7
}
8

9
// 修复：使用带缓冲 channel 或确保有接收者
10
func fixed1() {
11
    ch := make(chan int, 1)
12
    go func() {
13
        ch <- 42
14
    }()
15
    <-ch
16
}
17

18
// 场景2：select 缺少退出条件
19
func leak2() {
20
    ch := make(chan int)
21
    go func() {
22
        for {
23
            select {
24
            case v := <-ch:
25
                fmt.Println(v)
26
            // 缺少退出条件
27
            }
28
        }
29
    }()
30
}
31

32
// 修复：添加 context 取消
33
func fixed2(ctx context.Context) {
34
    ch := make(chan int)
35
    go func() {
36
        for {
37
            select {
38
            case v := <-ch:
39
                fmt.Println(v)
40
            case <-ctx.Done():
41
                return
42
            }
43
        }
44
    }()
45
}
46

47
// 场景3：阻塞在 nil channel
48
func leak3() {
49
    var ch chan int // nil channel
50
    go func() {
51
        <-ch // 永久阻塞
52
    }()
53
}
54

55
// 修复：确保 channel 已初始化
56
func fixed3() {
57
    ch := make(chan int)
58
    close(ch)
59
    go func() {
60
        <-ch // 接收零值后退出
61
    }()
62
}

1
import _ "net/http/pprof"
2

3
func main() {
4
    go func() {
5
        http.ListenAndServe("localhost:6060", nil)
6
    }()
7
    // 访问 http://localhost:6060/debug/pprof/goroutine?debug=1
8
    // 查看 goroutine 堆栈
9
}

1
// 错误：所有 goroutine 捕获同一个变量
2
func wrong() {
3
    for i := 0; i < 3; i++ {
4
        go func() {
5
            fmt.Println(i) // 可能输出 3, 3, 3
6
        }()
7
    }
8
}
9

10
// 正确：创建局部变量副本
11
func right() {
12
    for i := 0; i < 3; i++ {
13
        i := i // 捕获当前值
14
        go func() {
15
            fmt.Println(i) // 输出 0, 1, 2
16
        }()
17
    }
18
}
19

20
// 正确：作为参数传递
21
func alsoRight() {
22
    for i := 0; i < 3; i++ {
23
        go func(n int) {
24
            fmt.Println(n)
25
        }(i)
26
    }
27
}

1
// 错误：向已关闭 channel 发送会 panic
2
func wrong() {
3
    ch := make(chan int)
4
    close(ch)
5
    ch <- 1 // panic: send on closed channel
6
}
7

8
// 正确：使用 defer-recover 或确保只有一个关闭者
9
func right() {
10
    ch := make(chan int, 1)
11

12
    // 使用 sync.Once 确保只关闭一次
13
    var once sync.Once
14
    once.Do(func() { close(ch) })
15
}

1
// 错误：关闭 nil channel 会 panic
2
func wrong() {
3
    var ch chan int
4
    close(ch) // panic: close of nil channel
5
}
6

7
// 正确：检查 channel 是否为 nil
8
func right(ch chan int) {
9
    if ch != nil {
10
        close(ch)
11
    }
12
}

1
// 死锁场景
2
func deadlock() {
3
    ch1 := make(chan int)
4
    ch2 := make(chan int)
5

6
    go func() {
7
        <-ch1
8
        ch2 <- 1
9
    }()
10

11
    <-ch2 // 等待 ch2，但 ch2 需要 ch1 先被发送
12
    ch1 <- 1 // 永远无法到达
13
}
14

15
// 修复：使用缓冲 channel 或调整顺序
16
func fixed() {
17
    ch1 := make(chan int, 1)
18
    ch2 := make(chan int, 1)
19

20
    go func() {
21
        <-ch1
22
        ch2 <- 1
23
    }()
24

25
    ch1 <- 1
26
    <-ch2
27
}

1
// 竞态条件
2
func race() {
3
    var counter int
4
    var wg sync.WaitGroup
5

6
    for i := 0; i < 1000; i++ {
7
        wg.Add(1)
8
        go func() {
9
            defer wg.Done()
10
            counter++ // 竞态！
11
        }()
12
    }
13
    wg.Wait()
14
    fmt.Println(counter) // 结果不确定
15
}
16

17
// 使用互斥锁
18
func withMutex() {
19
    var counter int
20
    var mu sync.Mutex
21
    var wg sync.WaitGroup
22

23
    for i := 0; i < 1000; i++ {
24
        wg.Add(1)
25
        go func() {
26
            defer wg.Done()
27
            mu.Lock()
28
            counter++
29
            mu.Unlock()
30
        }()
31
    }
32
    wg.Wait()
33
}
34

35
// 使用 atomic
36
func withAtomic() {
37
    var counter int64
38
    var wg sync.WaitGroup
39

40
    for i := 0; i < 1000; i++ {
41
        wg.Add(1)
42
        go func() {
43
            defer wg.Done()
44
            atomic.AddInt64(&counter, 1)
45
        }()
46
    }
47
    wg.Wait()
48
}
49

50
// 使用 channel
51
func withChannel() {
52
    counter := make(chan int, 1)
53
    counter <- 0 // 初始化
54

55
    var wg sync.WaitGroup
56
    for i := 0; i < 1000; i++ {
57
        wg.Add(1)
58
        go func() {
59
            defer wg.Done()
60
            c := <-counter
61
            counter <- c + 1
62
        }()
63
    }
64
    wg.Wait()
65
    final := <-counter
66
    fmt.Println(final)
67
}

1
场景 ──▶ 推荐模式
2

3
数据需要多阶段处理 ──▶ Pipeline
4
多任务并行执行 ──▶ Fan-out/Fan-in
5
限制并发数量 ──▶ Worker Pool
6
需要取消/超时控制 ──▶ Context
7
并发任务需要错误处理 ──▶ errgroup