Posted inプログラミング

GoのSingleflightを極める：キャッシュスタンピードを防ぎ、アップストリームサービスを保護する

Programming tutorial - IT technology blog
Programming tutorial - IT technology blog

クイックスタート（5分）

ブラックフライデーの午前2時。商品サービスは秒間15,000リクエストを順調に処理し、セール対象の目玉モニターのデータを返しています。在庫データはTTL 5分でRedisに保持されています。しかし、そのキーが期限切れを迎えた瞬間、わずか50ミリ秒の間に2,500件のリクエストがAPIに殺到します。そのすべてがRedisでキャッシュミスとなり、まったく同一のクエリ（SELECT * FROM products WHERE id = 9812）がPostgreSQLへと直接雪崩れ込みます。

PostgresのCPU使用率は100%に張り付き、レイテンシは3msから8,500msへと急上昇。わずか4秒でコネクションプールの全128接続が枯渇し、クラスタ全体で504 Gateway Timeoutが連鎖的に発生します。これが本番環境で発生するスタンピード現象（Thundering Herd問題）の現実です。

Goは公式の準標準ライブラリ golang.org/x/sync/singleflight で、この問題に対するシンプルで直接的な解決策を提供しています。このパッケージは実行中（in-flight）の呼び出しを重複排除します。500件のリクエストが同一のリソースを同時に要求した場合でも、実行されるクエリは1つだけです。残りの499件の呼び出し元はその結果を待機し、共有します。

パッケージをモジュールに追加します：

go get golang.org/x/sync/singleflight

以下は、この動作を再現する実行可能な完全なコードです。5つのgoroutineがまったく同一のユーザーレコードを同時にクエリしますが、データベースからの取得処理は1回しか実行されません：

package main

import (
	"fmt"
	"sync"
	"sync/atomic"
	"time"

	"golang.org/x/sync/singleflight"
)

var dbQueryCount int32

func fetchUserDataFromDB(userID string) (string, error) {
	atomic.AddInt32(&dbQueryCount, 1)
	time.Sleep(150 * time.Millisecond) // DB I/Oのレイテンシをシミュレート
	return fmt.Sprintf("ユーザープロファイルデータ: %s", userID), nil
}

func main() {
	var group singleflight.Group
	var wg sync.WaitGroup
	userID := "user-4021"

	for i := 1; i <= 5; i++ {
		wg.Add(1)
		go func(workerID int) {
			defer wg.Done()
			v, err, shared := group.Do(userID, func() (interface{}, error) {
				return fetchUserDataFromDB(userID)
			})

			if err != nil {
				fmt.Printf("ワーカー %d が失敗しました: %v\n", workerID, err)
				return
			}

			fmt.Printf("ワーカー %d がデータを受信: %s (shared = %t)\n", workerID, v.(string), shared)
		}(i)
	}

	wg.Wait()
	fmt.Printf("\nDB上で実行されたクエリの合計数: %d\n", atomic.LoadInt32(&dbQueryCount))
}

このコードを実行してみてください。5つの呼び出しが競合しているにもかかわらず、fetchUserDataFromDB は1回しか実行されません。最初の呼び出し元が実際のデータ取得を行い、残りの4つは shared = true とともにまったく同一のペイロードを受け取ります。

内部の仕組み

このパッケージ全体は、コンパクトなステートマシンに基づいています。singleflight.Group の内部には、標準のミューテックスと非公開のハッシュマップが存在します：

type Group struct {
	mu sync.Mutex       // m を保護する
	m  map[string]*call // 遅延初期化される
}

アクティブな各クエリは、同期プリミティブ、重複カウンタ、戻り値のペイロードを保持する call 構造体によって追跡されます：

type call struct {
	wg   sync.WaitGroup
	val  interface{}
	err  error
	dups int
	chans []chan<- Result
}

goroutine A が group.Do("item-9812", fn) を呼び出した際の一連の処理フローは以下のとおりです：

  1. Singleflight はミューテックスをロックし、m["item-9812"] がすでに存在するか確認します。
  2. 存在しない場合、新しい *call を作成し、call.wg に 1 を加算してポインタをマップに保存した上で、ミューテックスを解放します。
  3. goroutine A がデータ取得コールバック fn() を実行します。

goroutine A がデータベースの I/O を待機している間に goroutine B と C が到着した場合：

  1. 両ルーチンはミューテックスを取得し、マップ内に m["item-9812"] がすでに存在することを確認します。
  2. call.dups をインクリメントし、ミューテックスをアンロックして、call.wg.Wait() で直接ブロックして待機します。

goroutine A の処理が完了すると、結果とエラーを call 構造体に書き込み、call.wg.Done() を呼び出し、再びミューテックスをロックしてマップからキーを削除した後にアンロックします。これにより goroutine B と C が起床し、共有された結果を取得します。たった1回のデータベースクエリで3つのコンシューマーにデータを提供できました。

本番環境での実践パターン

現実の本番サービスでは、単純な同期待ち以上の配慮が求められます。タイムアウト処理、ピンポイントでのキャッシュ無効化、防御的なメモリ安全性への対策が必要です。

DoChan を使ったコンテキストタイムアウトの処理

標準の group.Do() は内部関数が完了するまで処理をブロックします。もし上流のクライアントが接続を切断したり200msのHTTPタイムアウトを設定していたりすると、2秒かかる低速なデータベースクエリによって待機中のすべてのgoroutineが足止めされてしまいます。group.DoChan() を使用して、結果チャネルとリクエストコンテキストの双方を監視しましょう：

package cache

import (
	"context"
	"errors"
	"fmt"

	"golang.org/x/sync/singleflight"
)

type Service struct {
	sf singleflight.Group
}

func (s *Service) GetArticle(ctx context.Context, id string) (string, error) {
	resultChan := s.sf.DoChan(id, func() (interface{}, error) {
		return fetchFromOrigin(id)
	})

	select {
	case <-ctx.Done():
		// HTTPクライアントがリクエストを破棄した場合は早期リターン
		return "", ctx.Err()
	case res := <-resultChan:
		if res.Err != nil {
			return "", res.Err
		}
		val, ok := res.Val.(string)
		if !ok {
			return "", errors.New("singleflightの結果から予期しない型が返されました")
		}
		return val, nil
	}
}

func fetchFromOrigin(id string) (string, error) {
	return fmt.Sprintf("アイテム %s のデータ", id), nil
}

注意点として、最初の呼び出し元のコンテキストが期限切れになっても、バックグラウンドのコールバック処理は継続して実行されます。Singleflight は共有タスクを自動的にはキャンセルしないため、他の待機中呼び出し元は引き続きデータを正常に受け取ることができます。

group.Forget による低速・失敗キーの破棄

通常、singleflight はコールバックがリターンした後にのみキーを削除します。仮にサードパーティの決済ゲートウェイが30秒間ハングした場合、その間に到着したすべてのリクエストがスタックしたキューに滞留してしまいます。s.sf.Forget(id) を呼び出すことで、グループの内部マップから即座にそのキーを削除できます。

実行中だった元のワーカーはバックグラウンドで処理を継続しますが、直後に到着した呼び出し元はハングした処理の後ろに並ぶのではなく、新しいクエリを即座に開始します。これは素早いリトライが求められる障害発生時のエンドポイントで特に役立ちます。

変更可能なポインタの返却を避ける

並行処理のバグは、ポインタの共有によって混入することがよくあります。待機中のすべてのgoroutineはまったく同一の interface{} を受け取るため、*UserProfile のようなポインタを返すと、複数のスレッドが同一のメモリアドレスを同時に読み書きできてしまい、データ競合（Data Race）の原因となります。

値型、ディープコピー、またはシリアライズされた生バイト列（[]byte など）を常に返すようにしてください。呼び出し元ごとに独自のローカルコピーを安全にデシリアライズさせることが重要です。

実践から得られた運用の鉄則

高スループットな複数の本番クラスタでSingleflightを運用してきた経験から、システムの安定性を維持するために私たちが実践している4つの鉄則を紹介します：

  • SingleflightはDBの手前に置く防壁であり、キャッシュの代替ではない： 必ず最初にRedisやインメモリLRUキャッシュを確認してください。キャッシュミスが発生した場合にのみ singleflight.Group を経由してPostgreSQLから取得し、最新レコードをRedisに書き戻してからリターンします。
  • すべてのキーに名前空間を付与する： Singleflightは単純な文字列キーを使用します。もしユーザーのクエリと請求書のクエリがどちらもID "4501" を参照した場合、互いにブロックし合い、不正なペイロードを返すことになります。fmt.Sprintf("user:v1:%s", userID) のように厳格なプレフィックスを必ず付与してください。
  • グループのスコープをドメインごとに分離する： マイクロサービス全体で1つのグローバルな singleflight.Group を宣言してはいけません。各リポジトリ構造体の内部にインスタンスを持たせてスコープを限定します。これによりマップのロックが分離され、無関係なドメイン間でのミューテックス競合を排除できます。
  • ワーカーには常に独立したコンテキストタイムアウトを渡す： オリジンへのクエリにデッドラインが設定されていない場合、データベース接続がデッドロックすると待機しているすべての購読者がロックされてしまいます。取得関数内では、必ず明示的なクエリタイムアウト（例：context.WithTimeout(context.Background(), 3*time.Second)）を設定してください。

読み取り負荷の高いSQLテーブルやレートリミットのあるサードパーティAPIの手前にSingleflightを導入することで、突発的なトラフィック急増を確実に平滑化できます。データベースが重複したクエリでパンクすることなく、各リクエストを正確に1回だけ処理できるようになります。

Share:
Tags: