go-concurrency-web

Web 应用的 Go 并发

快速参考

主题	参考资料
Worker Pool 与 errgroup	references/worker-pools.md
速率限制	references/rate-limiting.md
竞态检测与修复	references/race-detection.md

核心规则

1. Goroutine 开销很小，但并非没有开销 — 每个 goroutine 会消耗约 2-8 KB 的栈空间。负载下无限制地创建 goroutine 会导致 OOM。
2. 始终提供关闭路径 — 启动的每个 goroutine 都必须有退出方式。使用 context.Context、关闭 channel 或 sync.WaitGroup。
3. 优先使用 channel 通信 — 使用 channel 协调 goroutine 之间的工作并发送完成信号。
4. 使用互斥锁保护状态 — 当 goroutine 共享可变状态时，使用 sync.Mutex、sync.RWMutex 或 sync/atomic 进行保护。
5. 切勿在 HTTP 处理器中直接启动 goroutine — 使用 Worker Pool、errgroup 或其他有界并发原语。

检查项（合并或审查前检查）

使用以下按顺序执行的检查来客观判定通过或失败；不要用“我在脑中验证过”来替代。

1. 竞态检测器
   • 对并发代码有变更的包运行 go test -race ./...，或对待测二进制文件运行 go build -race。
   • **通过条件：**退出代码为 0。如果报告“没有竞态”，请附上或引用 CI 输出／已保存的终端记录——没有该材料时不要声称代码不存在竞态。
2. 来自 HTTP 的后台工作有界
   • 检查启动请求 goroutine 之外工作的处理器和中间件。
   • **通过条件：**所有此类路径都使用有界原语（Worker Pool、注明容量的缓冲 channel、设置了明确并发上限的 errgroup），而不是针对每个传入请求无限制地执行 go。
3. 优雅拆除
   • 对启动了长期运行 goroutine 的进程，从关闭信号（或测试中的 defer）追踪到每组 goroutine 的 Wait()／关闭 channel／取消 context。
   • **通过条件：**能够指出调用链或证明关闭过程可以无挂起地完成的测试（没有遗留 goroutine）。

Worker Pool 模式

对于从 HTTP 处理器分发的后台任务，使用 Worker Pool。这样可以限制并发量并实现优雅关闭。

// Worker pool for background tasks (e.g., sending emails)
type WorkerPool struct {
    jobs   chan Job
    wg     sync.WaitGroup
    logger *slog.Logger
}

type Job struct {
    ID      string
    Execute func(ctx context.Context) error
}

func NewWorkerPool(numWorkers int, queueSize int, logger *slog.Logger) *WorkerPool {
    wp := &WorkerPool{
        jobs:   make(chan Job, queueSize),
        logger: logger,
    }

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

    return wp
}

func (wp *WorkerPool) worker(id int) {
    defer wp.wg.Done()
    for job := range wp.jobs {
        wp.logger.Info("processing job", "worker", id, "job_id", job.ID)
        if err := job.Execute(context.Background()); err != nil {
            wp.logger.Error("job failed", "worker", id, "job_id", job.ID, "err", err)
        }
    }
}

func (wp *WorkerPool) Submit(job Job) {
    wp.jobs <- job
}

func (wp *WorkerPool) Shutdown() {
    close(wp.jobs)
    wp.wg.Wait()
}

在 HTTP 处理器中使用

func (s *Server) handleCreateUser(w http.ResponseWriter, r *http.Request) {
    user, err := s.userService.Create(r.Context(), decodeUser(r))
    if err != nil {
        handleError(w, r, err)
        return
    }

    // Dispatch background task — never spawn raw goroutines in handlers
    s.workers.Submit(Job{
        ID: "welcome-email-" + user.ID,
        Execute: func(ctx context.Context) error {
            return s.emailService.SendWelcome(ctx, user)
        },
    })

    writeJSON(w, http.StatusCreated, user)
}

有关容量设置指南、背压、错误处理、重试模式，以及更简单的替代方案 errgroup，请参阅 references/worker-pools.md。

速率限制

使用 golang.org/x/time/rate 实现令牌桶速率限制。可将其用作中间件，实现全局限制或按 IP／用户限制。

要点：

• 全局速率限制可保护服务的整体承载能力
• 按 IP 进行速率限制可防止个别客户端独占资源
• 始终返回带有 Retry-After 标头的 429 Too Many Requests

有关中间件实现、按 IP 限制、清理过期限制器以及基于 API 密钥的限制，请参阅 references/rate-limiting.md。

竞态检测

在开发和 CI 中运行竞态检测器：

go test -race ./...
go build -race -o myserver ./cmd/server

竞态检测器可以捕获对共享内存的并发读写，但无法捕获逻辑竞态（例如 TOCTOU 错误）或死锁。

有关常见的 Web 处理器竞态、修复策略和 CI 集成，请参阅 references/race-detection.md。

处理器安全

每个传入的 HTTP 请求都在自己的 goroutine 中运行。服务器结构体上的任何共享可变状态都可能引发数据竞态。

// BAD — shared state without protection
type Server struct {
    requestCount int // data race!
}

func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
    s.requestCount++ // concurrent writes = race condition
}

// GOOD — use atomic or mutex
type Server struct {
    requestCount atomic.Int64
}

func (s *Server) handleRequest(w http.ResponseWriter, r *http.Request) {
    s.requestCount.Add(1)
}

// GOOD — use mutex for complex state
type Server struct {
    mu    sync.RWMutex
    cache map[string]*CachedItem
}

func (s *Server) handleGetCached(w http.ResponseWriter, r *http.Request) {
    s.mu.RLock()
    item, ok := s.cache[r.PathValue("key")]
    s.mu.RUnlock()
    // ...
}

处理器安全规则

• 请求范围的数据是安全的 — r.Context()、请求体和 URL 参数彼此隔离。
• 服务端结构体字段是共享的 — 处理器访问 *Server 上的任何字段时都需要同步。
• 数据库连接是安全的 — *sql.DB 会通过内部锁管理自己的连接池。
• 映射不安全 — 请使用 sync.Map，或通过互斥锁保护。
• 切片不安全 — 并发追加或读写时需要使用互斥锁。

反模式

无限制地创建 goroutine

// BAD — no limit on concurrent goroutines
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
    go func() {
        // What if 10,000 requests arrive at once?
        s.processWebhook(r.Context(), decodeWebhook(r))
    }()
    w.WriteHeader(http.StatusAccepted)
}

// GOOD — use a worker pool
func (s *Server) handleWebhook(w http.ResponseWriter, r *http.Request) {
    webhook := decodeWebhook(r)
    s.workers.Submit(Job{
        ID:      "webhook-" + webhook.ID,
        Execute: func(ctx context.Context) error {
            return s.processWebhook(ctx, webhook)
        },
    })
    w.WriteHeader(http.StatusAccepted)
}

忘记传递上下文

// BAD — loses cancellation signal
func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
    results, err := s.search(context.Background(), r.URL.Query().Get("q"))
    // ...
}

// GOOD — use request context
func (s *Server) handleSearch(w http.ResponseWriter, r *http.Request) {
    results, err := s.search(r.Context(), r.URL.Query().Get("q"))
    // ...
}

因缺少通道接收方而导致 goroutine 泄漏

// BAD — goroutine blocks forever if nobody reads the channel
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
    ch := make(chan *Response)
    go func() {
        resp, _ := http.Get(url) // blocks forever if ctx cancels
        ch <- resp               // stuck here if nobody reads
    }()
    select {
    case resp := <-ch:
        return resp, nil
    case <-ctx.Done():
        return nil, ctx.Err() // goroutine leaked!
    }
}

// GOOD — use buffered channel so goroutine can exit
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
    ch := make(chan *Response, 1) // buffered — goroutine can always send
    go func() {
        resp, _ := http.Get(url)
        ch <- resp
    }()
    select {
    case resp := <-ch:
        return resp, nil
    case <-ctx.Done():
        return nil, ctx.Err()
    }
}

使用 time.Sleep 进行协调

// BAD — sleeping to wait for goroutines
go doWork()
time.Sleep(5 * time.Second) // hoping it finishes

// GOOD — use sync primitives
var wg sync.WaitGroup
wg.Add(1)
go func() {
    defer wg.Done()
    doWork()
}()
wg.Wait()