Web应用的Go并发编程

快速参考

核心规则

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

工作池模式

使用工作池处理从HTTP处理器分发的后台任务。这可以限制并发并提供优雅关闭。

go
// 后台任务的工作池（例如发送邮件）
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(处理任务, worker, id, job_id, job.ID)
if err := job.Execute(context.Background()); err != nil {
wp.logger.Error(任务失败, 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处理器中的使用

go
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
}

// 分发后台任务 — 绝不在处理器中直接创建goroutine
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)
}

参见references/worker-pools.md了解容量规划、背压处理、错误处理、重试模式以及作为更简单替代方案的errgroup。

速率限制

使用golang.org/x/time/rate实现令牌桶速率限制。作为中间件应用于全局限制或按IP/按用户限制。

关键点：

• - 全局速率限制保护整体服务容量
• 按IP速率限制防止单个客户端独占资源
• 始终返回429 Too Many Requests并附带Retry-After头

参见references/rate-limiting.md了解中间件实现、按IP限制、过期限制器清理和基于API密钥的限制。

竞态检测

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

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

竞态检测器捕获对共享内存的并发读写。它不会捕获逻辑竞态（例如TOCTOU错误）或死锁。

参见references/race-detection.md了解常见的Web处理器竞态、修复策略和CI集成。

处理器安全

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

go
// 错误 — 无保护的共享状态
type Server struct {
requestCount int // 数据竞态！
}

func (s Server) handleRequest(w http.ResponseWriter, r http.Request) {
s.requestCount++ // 并发写入 = 竞态条件
}

// 正确 — 使用原子操作或互斥锁
type Server struct {
requestCount atomic.Int64
}

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

// 正确 — 对复杂状态使用互斥锁
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

go
// 错误 — 并发goroutine无限制
func (s Server) handleWebhook(w http.ResponseWriter, r http.Request) {
go func() {
// 如果同时收到10,000个请求怎么办？
s.processWebhook(r.Context(), decodeWebhook(r))
}()
w.WriteHeader(http.StatusAccepted)
}

// 正确 — 使用工作池
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)
}

忘记传播上下文

go
// 错误 — 丢失取消信号
func (s Server) handleSearch(w http.ResponseWriter, r http.Request) {
results, err := s.search(context.Background(), r.URL.Query().Get(q))
// ...
}

// 正确 — 使用请求上下文
func (s Server) handleSearch(w http.ResponseWriter, r http.Request) {
results, err := s.search(r.Context(), r.URL.Query().Get(q))
// ...
}

缺少通道接收者导致的goroutine泄漏

go
// 错误 — 如果没有读取通道，goroutine会永远阻塞
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
ch := make(chan *Response)
go func() {
resp, _ := http.Get(url) // 如果ctx取消则永远阻塞
ch <- resp // 如果没人读取则卡在这里
}()
select {
case resp := <-ch:
return resp, nil
case <-ctx.Done():
return nil, ctx.Err() // goroutine泄漏！
}
}

// 正确 — 使用缓冲通道以便goroutine可以退出
func fetchWithTimeout(ctx context.Context, url string) (*Response, error) {
ch := make(chan *Response, 1) // 缓冲 — goroutine总是可以发送
go func() {
resp, _ := http.Get(url)
ch <- resp
}()
select {
case resp := <-ch:
return resp, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}

使用time.Sleep进行协调

go
// 错误 — 通过睡眠等待goroutine
go doWork()
time.Sleep(5 * time.Second) // 希望它完成

// 正确 — 使用同步原语
var wg sync.WaitGroup
wg.Add(1)
go func() {
defer wg.Done()
doWork()
}()
wg.Wait()