context 与并发模式

第 10 章 · context、Worker Pool 与超时取消

本章目标：掌握 context.Context 的取消、超时与传值；实现可优雅退出的 Worker Pool；用 semaphore（chan struct{}）限流；在 HTTP 客户端传递 context（衔接 gin-web）；改造 toolkit-go 解析任务，支持 Ctrl+C 与超时取消。

学时建议：3～4 小时（含 1.5 小时跟练）

前置：完成 go-dev ch09（goroutine、channel、WaitGroup）。

10.1 场景说明：批处理必须能「停下来」

toolkit-go 在解析 百万行 access 日志时，用户可能：

• 按 Ctrl+C 终止进程；
• 通过 -timeout 30s 限制单次批处理时长；
• 在 gin-web 中，HTTP 客户端断开时，服务端应停止无效计算。

context 是 Go 标准库定义的 取消信号载体，从 context.Background() 根节点派生，沿调用链传递：

main / HTTP Handler
      │
      ▼
 context.WithCancel / WithTimeout(parent, 30s)
      │
      ├──► Worker Pool goroutine × N  ──► select { case <-ctx.Done(): return }
      │
      └──► http.NewRequestWithContext(ctx, ...)

Gin 对照：c.Request.Context() 在客户端断开时会 Done，gin-web ch02+ 中间件可从中取 request_id。

10.2 context 基础 API

package main

import (
    "context"
    "fmt"
    "time"
)

func main() {
    // 根 context — 永不取消（除非进程退出）
    root := context.Background()
    _ = context.TODO() // 占位，重构时用

    // 手动取消
    ctx, cancel := context.WithCancel(root)
    defer cancel() // **惯例：WithXxx 返回的 cancel 必须 defer**

    go func() {
        <-ctx.Done()
        fmt.Println("worker stopped:", ctx.Err()) // context canceled
    }()
    cancel() // 通知所有监听 ctx.Done() 的 goroutine

    // 超时取消
    ctx2, cancel2 := context.WithTimeout(root, 3*time.Second)
    defer cancel2()
    select {
    case <-time.After(5 * time.Second):
        fmt.Println("should not reach")
    case <-ctx2.Done():
        fmt.Println("timeout:", ctx2.Err()) // context deadline exceeded
    }
}

铁律：WithCancel / WithTimeout / WithDeadline 返回的 cancel 必须调用（通常 defer cancel()），否则 timer 等资源泄漏。

10.3 Worker Pool：可取消版

在 ch09 基础上，worker 内 select 监听 ctx.Done()：

// internal/logparser/pool.go
package logparser

import (
    "context"
    "encoding/json"
    "sync"

    "example.com/toolkit-go/internal/model"
)

type PoolResult struct {
    Entry model.LogEntry
    Err   error // 单行解析错误可选上报
}

func RunParsePool(
    ctx context.Context,
    lines <-chan string,
    workers int,
    thresholdMs int,
) <-chan PoolResult {
    out := make(chan PoolResult, workers*2)
    var wg sync.WaitGroup

    worker := func() {
        defer wg.Done()
        for {
            select {
            case <-ctx.Done():
                return
            case line, ok := <-lines:
                if !ok {
                    return
                }
                var e model.LogEntry
                if err := json.Unmarshal([]byte(line), &e); err != nil {
                    continue
                }
                if e.DurationMs >= thresholdMs {
                    select {
                    case out <- PoolResult{Entry: e}:
                    case <-ctx.Done():
                        return
                    }
                }
            }
        }
    }

    wg.Add(workers)
    for i := 0; i < workers; i++ {
        go worker()
    }

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

调用方：

ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()

jobs := make(chan string, 100)
go func() {
    defer close(jobs)
    for _, ln := range allLines {
        select {
        case jobs <- ln:
        case <-ctx.Done():
            return
        }
    }
}()

results := RunParsePool(ctx, jobs, 4, 500)
for r := range results {
    _ = r.Entry
}

10.4 semaphore 限流

限制 同时 运行的 goroutine 数（如并发 HTTP 抓取、并发打开文件数）：

func FetchPaths(ctx context.Context, paths []string, limit int) error {
    if limit <= 0 {
        limit = 8
    }
    sem := make(chan struct{}, limit) // 空 struct 不占额外内存
    var wg sync.WaitGroup
    errCh := make(chan error, 1)

    for _, p := range paths {
        select {
        case <-ctx.Done():
            return ctx.Err()
        default:
        }

        sem <- struct{}{} // 获取令牌；满则阻塞
        wg.Add(1)
        go func(path string) {
            defer wg.Done()
            defer func() { <-sem }() // 释放令牌

            if err := processOneLogFile(ctx, path); err != nil {
                select {
                case errCh <- err:
                default:
                }
            }
        }(p)
    }

    wg.Wait()
    select {
    case err := <-errCh:
        return err
    default:
        return nil
    }
}