キャッシュ — Redisによるホットクエリの高速化

Q: ブルームフィルターはどう実装しますか?

redisbloom モジュールまたはPythonの pybloom_live ライブラリを使用します。既存のすべてのプロダクトIDをブルームフィルターに追加し, 各クエリ前にチェックします。偽陽性率は約1%で許容範囲です。

Q: キャッシュヒット率はどうモニタリングしますか?

Redisの INFO stats コマンドが keyspace_hits と keyspace_misses を返します。ヒット率 = hits / (hits + misses) で計算します。Prometheus + Grafanaモニタリングダッシュボードに統合します。

キャッシュは冷蔵庫のようなものです。よく使う食材 (ホットデータ)は冷蔵庫に常備してすぐに手に取れるようにし, 毎回スーパー (データベース)に行かなくて済むようにします。ただし食材には期限 (TTL)があり, 入荷時には更新が必要です (キャッシュ無効化)。

1. 学ぶ内容

Redis接続管理:redis-py非同期クライアントとFastAPIの統合
キャッシュ戦略:Cache-Asideパターン, TTL期限切れ, キャッシュキー命名設計
キャッシュ無効化:書き込み時の能動的無効化, バッチ価格更新時のキャッシュ整合性
キャッシュ穿透/突破/雪崩保護:ブルームフィルター, ミューテックスロック, ランダムTTL
Aliceシナリオ:人気プロダクトの価格キャッシュ — P99レイテンシが200msから5msに削減

2. Aliceのリアルストーリー

(1) ペインポイント:人気クエリがデータベースを圧迫

PriceTrackerには100の人気プロダクトがあり, クエリ量の80%を占め, 1秒間に8,000クエリがあります。すべてのクエリがPostgreSQLを経由するため, データベースCPUが90%に跳ね上がり, P99レイテンシが200msに達しています。Charlieはデータベースインスタンスの追加で月500USDかかると言いますが, 80%のクエリ結果は5分以内に変わりません。

(2) Redisキャッシュのソリューション

人気プロダクトの価格データをRedisにキャッシュします。クエリ時はまずRedisを確認 (5ms), ヒットしなければPostgreSQLをクエリ (50ms)し, 結果を5分TTLでRedisに書き戻します。

PYTHON

async def get_product_cached(product_id: int, db, redis):
    cached = await redis.get(f"product:{product_id}")
    if cached:
        return json.loads(cached)
    product = await db.execute(select(Product).where(Product.id == product_id))
    data = product.scalar_one_or_none()
    await redis.setex(f"product:{product_id}", 300, json.dumps(data))
    return data

(3) 収益

人気クエリの80%がRedisキャッシュにヒットし, P99レイテンシが200msから5msに低下, データベースQPSが8,000から1,600に減少し, Charlieのデータベースインスタンスが4から1に削減され, 月1,500USDの節約になりました。

3. Redis接続管理

(1) 非同期クライアントの統合

(1) ▶サンプル:Redis接続依存性

PYTHON

import redis.asyncio as aioredis
from fastapi import FastAPI, Depends

REDIS_URL = "redis://localhost:6379/0"

# アプリケーションレベルのRedis接続プール
redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)

async def get_redis() -> aioredis.Redis:
    """接続プールからのリクエストスコープRedisクライアント"""
    client = aioredis.Redis(connection_pool=redis_pool)
    try:
        yield client
    finally:
        await client.aclose()

# 起動/シャットダウンライフサイクル
app = FastAPI()

@app.on_event("startup")
async def startup():
    global redis_pool
    redis_pool = aioredis.ConnectionPool.from_url(REDIS_URL, max_connections=20)

@app.on_event("shutdown")
async def shutdown():
    await redis_pool.disconnect()

出力:

TEXT

# 関数定義成功

(2) Redisデータ構造の選択

構造	コマンド	用途	PriceTrackerの用途
String	SET/GET	単一値キャッシュ	プロダクト価格キャッシュ
Hash	HSET/HGET	オブジェクトキャッシュ	プロダクト詳細キャッシュ
List	LPUSH/LRANGE	時系列	価格変更履歴
Set	SADD/SMEMBERS	重複排除セット	プッシュ済み価格ID
ZSet	ZADD/ZRANGE	ランキング	売れ筋プロダクト

4. Cache-Asideモード

(1) 読み書きプロセス

flowchart TD
    Request[読み取りリクエスト] --> CheckRedis{キャッシュヒット?}
    CheckRedis -->|はい| Return[キャッシュデータを返す]
    CheckRedis -->|いいえ| QueryDB[データベースをクエリ]
    QueryDB --> WriteCache[TTL付きでRedisに書き込み]
    WriteCache --> Return2[データを返す]
    
    WriteRequest[書き込みリクエスト] --> UpdateDB[データベースを更新]
    UpdateDB --> Invalidate[キャッシュを無効化]
    Invalidate --> Return3[成功を返す]

(1) ▶サンプル:Cache-Asideプロダクトクエリ

PYTHON

import json
from fastapi import FastAPI, Depends
from redis.asyncio import Redis

app = FastAPI()

@app.get("/api/v1/products/{product_id}")
async def get_product(
    product_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    # ステップ1:Redisキャッシュを確認
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # ステップ2:キャッシュミス - データベースをクエリ
    repo = ProductRepository(db)
    product = await repo.get_by_id(product_id)
    if not product:
        raise HTTPException(status_code=404, detail="Product not found")
    
    # ステップ3:TTL付きでキャッシュに書き込み
    product_data = ProductResponse.model_validate(product).model_dump()
    await redis.setex(cache_key, 300, json.dumps(product_data))  # 5分TTL
    
    return product_data

出力:

TEXT

# 関数定義成功

(2) ▶サンプル:書き込み時の能動的無効化

PYTHON

@app.put("/api/v1/products/{product_id}")
async def update_product(
    product_id: int,
    update: ProductUpdate,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
    user=Depends(get_current_user),
):
    repo = ProductRepository(db)
    product = await repo.update(product_id, update.model_dump(exclude_unset=True))
    
    # 書き込み後にキャッシュを無効化
    await redis.delete(f"product:{product_id}")
    
    return ProductResponse.model_validate(product).model_dump()

出力:

TEXT

# 関数定義成功

(2) キャッシュキー命名規則

パターン	キー名	TTL	説明
単一キャッシュ	`product:{id}`	5分	単一プロダクト詳細
リストキャッシュ	`products:cat:{category}:p:{page}`	2分	ページネーションリスト
カウンター	`count:products:cat:{category}`	1分	プロダクト数
人気ランキング	`ranking:products:hot`	10分	ZSetランキング
レート制限カウント	`ratelimit:{ip}`	60秒	APIレート制限

5. キャッシュ穿透/突破/雪崩保護

(1) 3大キャッシュ問題

問題	原因	対策
穿透	存在しないデータの検索でキャッシュをバイパスしてDBに直接アクセス	ブルームフィルター, Null値キャッシュ
突破	ホットキーの期限切れ瞬間に大量リクエストがデータベースになだれ込む	ミューテックスロック, 期限なしキー + 非同期更新
雪崩	キーの一斉期限切れ	ランダムTTLオフセット

(1) ▶サンプル:Null値キャッシュで穿透を防止

PYTHON

@app.get("/api/v1/products/{product_id}")
async def get_product_with_null_cache(
    product_id: int,
    db: AsyncSession = Depends(get_db),
    redis: Redis = Depends(get_redis),
):
    cache_key = f"product:{product_id}"
    cached = await redis.get(cache_key)
    
    if cached:
        data = json.loads(cached)
        if data.get("_null"):
            raise HTTPException(status_code=404, detail="Product not found")
        return data
    
    # DBをクエリ
    repo = ProductRepository(db)
    product = await repo.get_by_id(product_id)
    
    if not product:
        # 短いTTLでNull値をキャッシュして穿透を防止
        await redis.setex(cache_key, 60, json.dumps({"_null": True}))
        raise HTTPException(status_code=404, detail="Product not found")
    
    product_data = ProductResponse.model_validate(product).model_dump()
    await redis.setex(cache_key, 300, json.dumps(product_data))
    return product_data

出力:

TEXT

# 関数定義成功

(2) ▶サンプル:ミューテックスロックで突破を防止

PYTHON

import asyncio

async def get_product_with_lock(
    product_id: int,
    db: AsyncSession,
    redis: Redis,
):
    cache_key = f"product:{product_id}"
    lock_key = f"lock:product:{product_id}"
    
    cached = await redis.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # ロックの取得を試行 (1つのリクエストのみがキャッシュを再構築)
    lock_acquired = await redis.set(lock_key, "1", nx=True, ex=10)
    
    if lock_acquired:
        try:
            # このリクエストがキャッシュを再構築
            repo = ProductRepository(db)
            product = await repo.get_by_id(product_id)
            if product:
                data = ProductResponse.model_validate(product).model_dump()
                await redis.setex(cache_key, 300, json.dumps(data))
                return data
        finally:
            await redis.delete(lock_key)
    else:
        # 他のリクエストは少し待ってからキャッシュをリトライ
        await asyncio.sleep(0.1)
        cached = await redis.get(cache_key)
        if cached:
            return json.loads(cached)
    
    raise HTTPException(status_code=404, detail="Product not found")

出力:

TEXT

# 関数定義成功

(3) ▶サンプル:ランダムTTLで雪崩を防止

PYTHON

import random

def get_cache_ttl(base_ttl: int = 300, jitter: int = 60) -> int:
    """TTLにランダムジッターを追加してキャッシュ雪崩を防止"""
    return base_ttl + random.randint(-jitter, jitter)

# 使用例
await redis.setex(cache_key, get_cache_ttl(300, 60), json.dumps(data))
# TTL:240-360秒 (5分 ± 1分ジッター)

出力:

TEXT

# 関数定義成功

(2) キャッシュ整合性ポリシーの比較

戦略	書き込み操作	利点	欠点
Cache-Aside	先にDB更新, 次にキャッシュクリア	シンプルで信頼性が高い	短い不一致の窓
Write-Through	キャッシュとDBに同時書き込み	高い整合性	書き込みレイテンシが高い
Write-Behind	先にキャッシュに書き込み, 次に非同期でDBに	書き込みパフォーマンスが良い	データ損失の可能性

❓ よくある質問

Q キャッシュにはJSONとPickleのどちらを保存すべきですか?

A JSONを推奨します。人間が読め, 言語横断的で安全です (Pickleにはデシリアライゼーション脆弱性があります)。orjsonはjsonより3-10倍高速です。

Q 適切なTTL設定はどう決めるべきですか?

A データの変更頻度によります。PriceTrackerの価格データは5分ごとに更新 (短い不一致を許容), プロダクト詳細は30分, ランキングは10分です。雪崩防止のためにランダムオフセットを追加します。

Q Redis接続プールサイズはどう設定すべきですか?

A FastAPI Worker数に合わせます。各Workerは5-10のRedis接続が必要です。4 Worker x 10 = 40接続, 余裕を見てmax_connectionsを50に設定します。

Q ブルームフィルターはどう実装しますか?

A redisbloomモジュールまたはPythonのpybloom_liveライブラリを使用します。既存のすべてのプロダクトIDをブルームフィルターに追加し, 各クエリ前にチェックします。偽陽性率は約1%で許容範囲です。

Q キャッシュとデータベースの間に不一致がある場合はどうすればよいですか?

A Cache-Asideモードでは一時的な不一致は正常です。戦略:先にDBを更新し, 次にキャッシュを削除 (キャッシュの更新ではなく), 遅延二重削除 (キャッシュ削除 → DB更新 → 遅延後再削除)で不一致の窓を最小化します。

Q キャッシュヒット率はどうモニタリングしますか?

A RedisのINFO statsコマンドがkeyspace_hitsとkeyspace_missesを返します。ヒット率 = hits / (hits + misses) で計算します。Prometheus + Grafanaモニタリングダッシュボードに統合します。

📖 まとめ

Redis非同期クライアント (redis.asyncio)+ 接続プール + yield依存性注入でFastAPIと統合
Cache-Asideモード:読み取り時はキャッシュを先に確認, ヒットしなければDBをクエリしてキャッシュに書き込み, 書き込み時はDBを更新してキャッシュエントリを削除
キャッシュキーは{entity}:{id}パターンで命名し, シナリオに応じて異なるTTLを設定
Null値キャッシュで穿透防止, ミューテックスロックで突破防止, ランダムTTLで雪崩防止の3大防御戦略
PriceTracker:人気プロダクトのキャッシュヒット率80%, P99レイテンシが200msから5msに改善

📝 練習問題

基本問題 (難易度 ⭐):Redis非同期クライアント依存性を設定し, シンプルなGET/SETキャッシュエンドポイントを実装して, データの保存と取得を検証します。ヒント:redis.asyncio + Depends(get_redis) + setex()
応用問題 (難易度 ⭐⭐):PriceTrackerのプロダクト検索にCache-Asideパターンを実装します。キャッシュから読み取り → キャッシュミスならDBをクエリ → キャッシュに書き込み (TTL 5分), 書き込み操作でDB更新後にキャッシュエントリを削除します。ヒント:await redis.get(key) + await redis.delete(key)
チャレンジ (難易度 ⭐⭐⭐):包括的なキャッシュ保護を実装します。Null値キャッシュの穿透防止 (短いTTL 60s), ミューテックスロックの突破防止 (SET lock NX EX), ランダムTTLの雪崩防止 (300s ± 60s)を行い, 保護ロジックが正しいことをテストで検証します。ヒント:redis.set(lock_key, "1", nx=True, ex=10) + random.randint()

---|

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

私たちについて

404 Not Found

キャッシュ — Redisによるホットクエリの高速化

1. 学ぶ内容

2. Aliceのリアルストーリー

(1) ペインポイント:人気クエリがデータベースを圧迫

(2) Redisキャッシュのソリューション

(3) 収益

3. Redis接続管理

(1) 非同期クライアントの統合

(1) ▶サンプル:Redis接続依存性

(2) Redisデータ構造の選択

4. Cache-Asideモード

(1) 読み書きプロセス

(1) ▶サンプル:Cache-Asideプロダクトクエリ

(2) ▶サンプル:書き込み時の能動的無効化

(2) キャッシュキー命名規則

5. キャッシュ穿透/突破/雪崩保護

(1) 3大キャッシュ問題

(1) ▶サンプル:Null値キャッシュで穿透を防止

(2) ▶サンプル:ミューテックスロックで突破を防止

(3) ▶サンプル:ランダムTTLで雪崩を防止

(2) キャッシュ整合性ポリシーの比較

❓ よくある質問

📖 まとめ

📝 練習問題