go-concurrency-web

Web アプリケーションの Go 並行処理

クイックリファレンス

トピック	リファレンス
ワーカープールと errgroup	references/worker-pools.md
レート制限	references/rate-limiting.md
データ競合の検出と修正	references/race-detection.md

基本ルール

1. ゴルーチンは軽量ですが、無料ではありません — 各ゴルーチンは約 2～8 KB のスタックを消費します。負荷の高い状況で無制限に生成すると、OOM につながります。
2. 必ずシャットダウン経路を用意してください — 起動したすべてのゴルーチンには、終了する方法が必要です。context.Context、チャネルのクローズ、または sync.WaitGroup を使用してください。
3. 通信にはチャネルを優先してください — ゴルーチン間の作業を調整し、完了を通知するにはチャネルを使用してください。
4. 状態の保護にはミューテックスを使用してください — ゴルーチン間で可変状態を共有する場合は、sync.Mutex、sync.RWMutex、または sync/atomic で保護してください。
5. HTTP ハンドラー内で直接ゴルーチンを起動しないでください — ワーカープール、errgroup、またはその他の並行処理数を制限するプリミティブを使用してください。

ゲート（マージまたはレビューの前に確認）

客観的に合否を判断できるよう、以下の順序付きチェックを使用してください。「頭の中で確認した」に置き換えないでください。

1. データ競合検出器
   • 並行処理コードを変更したパッケージでは go test -race ./... を実行するか、テスト対象のバイナリでは go build -race を実行してください。
   • 合格： 終了コードが 0 であること。「データ競合なし」と報告する場合は、CI 出力または保存したターミナルの記録を添付または引用してください。その証拠なしに問題がないと断言しないでください。
2. HTTP から起動するバックグラウンド処理の制限
   • リクエストを処理するゴルーチンの外で処理を開始するハンドラーとミドルウェアを確認してください。
   • 合格： 該当するすべての経路で、並行処理数を制限するプリミティブ（ワーカープール、容量が明記されたバッファ付きチャネル、明示的な並行処理数上限を設けた errgroup）を使用していること。受信リクエストごとに無制限に go を起動してはいけません。
3. 正常な終了処理
   • 長時間稼働するゴルーチンを起動するプロセスでは、シャットダウンシグナル（またはテストの defer）から、それぞれのゴルーチン群に対する Wait()、チャネルのクローズ、または context のキャンセルまでを追跡してください。
   • 合格： ハングせずにシャットダウンが完了することを証明する呼び出しチェーンまたはテストを示せること（孤立したゴルーチンがないこと）。

ワーカープールパターン

HTTP ハンドラーからディスパッチするバックグラウンドタスクには、ワーカープールを使用してください。これにより並行処理数を制限し、正常なシャットダウンが可能になります。

// 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 リクエストはそれぞれ独自のゴルーチンで実行されます。サーバー構造体上の共有可変状態はすべて、データ競合が発生する可能性があります。

// 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 を使用するか、ミューテックスで保護してください。
• スライスは安全ではありません — 並行して追加または読み書きするには、ミューテックスが必要です。

アンチパターン

上限を設けずにゴルーチンを生成する

// 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"))
    // ...
}

チャネルの受信側がないことによるゴルーチンリーク

// 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()