redis-patterns

Redis パターン

一般的なバックエンドユースケースにおけるRedisのベストプラクティスに関するクイックリファレンス。

仕組み

Redisはインメモリデータ構造ストアで、文字列、ハッシュ、リスト、セット、ソート済みセット、ストリームなどをサポートします。個々のRedisコマンドは単一インスタンス上でアトミックです。マルチステップのワークフローでは、アトミック性を維持するためにLuaスクリプト、MULTI/EXECトランザクション、または明示的な同期が必要です。データはRDBスナップショットまたはAOFログを介してオプションで永続化されます。クライアントはTCP上でRESPプロトコルを使用して通信します。リクエストごとのハンドシェイクオーバーヘッドを避けるために、コネクションプールは必須です。

アクティブ化のタイミング

• アプリケーションへのキャッシュ追加
• レート制限またはスロットリングの実装
• 分散ロックまたは調整機能の構築
• セッションまたはトークンストレージのセットアップ
• メッセージングのためのPub/SubまたはRedis Streamsの使用
• Redisの本番環境設定（プーリング、エビクション、クラスタリング）

データ構造チートシート

ユースケース	構造	キーの例
シンプルなキャッシュ	文字列	product:123
ユーザーセッション	ハッシュ	session:abc
リーダーボード	ソート済みセット	scores:weekly
ユニークビジター	セット	visitors:2024-01-01
アクティビティフィード	リスト	feed:user:456
イベントストリーム	ストリーム	events:orders
カウンター / レート制限	文字列 (INCR)	ratelimit:user:123
Bloomフィルター / HLL	HyperLogLog	hll:pageviews

コアパターン

キャッシュアサイド（遅延ロード）

import redis
import json

r = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_product(product_id: int):
    cache_key = f"product:{product_id}"
    cached = r.get(cache_key)

    if cached:
        return json.loads(cached)

    product = db.query("SELECT * FROM products WHERE id = %s", product_id)
    r.setex(cache_key, 3600, json.dumps(product))  # TTL: 1時間
    return product

ライトスルーキャッシュ

def update_product(product_id: int, data: dict):
    # まずDBに書き込み
    db.execute("UPDATE products SET ... WHERE id = %s", product_id)

    # 直ちにキャッシュを更新
    cache_key = f"product:{product_id}"
    r.setex(cache_key, 3600, json.dumps(data))

キャッシュ無効化

# タグベースの無効化 — 関連するキーをセットでグルーピング
def cache_product(product_id: int, category_id: int, data: dict):
    key = f"product:{product_id}"
    tag = f"tag:category:{category_id}"
    pipe = r.pipeline(transaction=True)
    pipe.setex(key, 3600, json.dumps(data))
    pipe.sadd(tag, key)
    pipe.expire(tag, 3600)
    pipe.execute()

def invalidate_category(category_id: int):
    tag = f"tag:category:{category_id}"
    keys = r.smembers(tag)
    if keys:
        r.delete(*keys)
    r.delete(tag)

セッションストレージ

import time
import uuid

def create_session(user_id: int, ttl: int = 86400) -> str:
    session_id = str(uuid.uuid4())
    key = f"session:{session_id}"
    pipe = r.pipeline(transaction=True)
    pipe.hset(key, mapping={
        "user_id": user_id,
        "created_at": int(time.time()),
    })
    pipe.expire(key, ttl)
    pipe.execute()
    return session_id

def get_session(session_id: str) -> dict | None:
    data = r.hgetall(f"session:{session_id}")
    return data if data else None

def delete_session(session_id: str):
    r.delete(f"session:{session_id}")

レート制限

固定ウィンドウ（シンプル）

def is_rate_limited(user_id: int, limit: int = 100, window: int = 60) -> bool:
    key = f"ratelimit:{user_id}:{int(time.time()) // window}"
    pipe = r.pipeline(transaction=True)
    pipe.incr(key)
    pipe.expire(key, window)
    count, _ = pipe.execute()
    return count > limit

スライディングウィンドウ（Lua — アトミック）

-- sliding_window.lua
local key = KEYS[1]
local now = tonumber(ARGV[1])
local window = tonumber(ARGV[2])
local limit = tonumber(ARGV[3])

redis.call('ZREMRANGEBYSCORE', key, 0, now - window)
local count = redis.call('ZCARD', key)

if count < limit then
    -- 一意なメンバー（現在時刻+シーケンス）を使用して、同一ミリ秒内の衝突を回避
    local seq_key = key .. ':seq'
    local seq = redis.call('INCR', seq_key)
    redis.call('EXPIRE', seq_key, math.ceil(window / 1000))
    redis.call('ZADD', key, now, now .. '-' .. seq)
    redis.call('EXPIRE', key, math.ceil(window / 1000))
    return 1
end
return 0

sliding_window = r.register_script(open('sliding_window.lua').read())

def allow_request(user_id: int) -> bool:
    key = f"ratelimit:sliding:{user_id}"
    now = int(time.time() * 1000)
    return bool(sliding_window(keys=[key], args=[now, 60000, 100]))

分散ロック

分散ロック（シングルノード — SET NX PX）

import uuid

def acquire_lock(resource: str, ttl_ms: int = 5000) -> str | None:
    lock_key = f"lock:{resource}"
    token = str(uuid.uuid4())
    acquired = r.set(lock_key, token, px=ttl_ms, nx=True)
    return token if acquired else None

def release_lock(resource: str, token: str) -> bool:
    release_script = """
    if redis.call('get', KEYS[1]) == ARGV[1] then
        return redis.call('del', KEYS[1])
    else
        return 0
    end
    """
    result = r.eval(release_script, 1, f"lock:{resource}", token)
    return bool(result)

# 使用例
token = acquire_lock("order:payment:123")
if token:
    try:
        process_payment()
    finally:
        release_lock("order:payment:123", token)

マルチノードセットアップには、完全なRedlockアルゴリズムを実装した redlock-py ライブラリを使用してください。

Pub/Sub と ストリーム

Pub/Sub（ファイアアンドフォーゲット）

# パブリッシャー
def publish_event(channel: str, payload: dict):
    r.publish(channel, json.dumps(payload))

# サブスクライバー（ブロッキング — 別スレッド/プロセスで実行）
def subscribe_events(channel: str):
    pubsub = r.pubsub()
    pubsub.subscribe(channel)
    for message in pubsub.listen():
        if message['type'] == 'message':
            handle(json.loads(message['data']))

Redis Streams（永続的なキュー）

# プロデューサー
def emit(stream: str, event: dict):
    r.xadd(stream, event, maxlen=10000)  # ストリーム長を制限

# コンシューマグループ — 最低1回の配信を保証
try:
    r.xgroup_create('events:orders', 'processor', id='0', mkstream=True)
except Exception:
    pass  # グループは既に存在

def consume(stream: str, group: str, consumer: str):
    while True:
        messages = r.xreadgroup(group, consumer, {stream: '>'}, count=10, block=2000)
        for _, entries in (messages or []):
            for msg_id, data in entries:
                process(data)
                r.xack(stream, group, msg_id)

配信保証、コンシューマグループ、リプレイが必要な場合は、Pub/Subよりも Streams を推奨します。

キーデザイン

命名規則

# パターン: resource:id:field
user:123:profile
order:456:status
cache:product:789

# パターン: namespace:resource:id
myapp:session:abc123
myapp:ratelimit:user:123

# パターン: resource:date（時間バインドされたキー）
stats:pageviews:2024-01-01

TTL戦略

データタイプ	推奨TTL
ユーザーセッション	24時間 (86400)
API応答キャッシュ	5～15分
レート制限ウィンドウ	ウィンドウサイズに合わせる
短命トークン	5～10分
リーダーボード	1時間～24時間
静的/参照データ	1時間～1週間

常にTTLを設定してください。TTLなしのキーは無制限に蓄積され、メモリプレッシャーを引き起こします。

接続管理

コネクションプーリング

from redis import ConnectionPool, Redis

pool = ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=20,
    decode_responses=True,
    socket_connect_timeout=2,
    socket_timeout=2,
)

r = Redis(connection_pool=pool)

クラスタモード

from redis.cluster import RedisCluster

r = RedisCluster(
    startup_nodes=[{"host": "redis-1", "port": 6379}],
    decode_responses=True,
    skip_full_coverage_check=True,
)

センチネル（高可用性）

from redis.sentinel import Sentinel

sentinel = Sentinel(
    [('sentinel-1', 26379), ('sentinel-2', 26379)],
    socket_timeout=0.5,
)
master = sentinel.master_for('mymaster', decode_responses=True)
replica = sentinel.slave_for('mymaster', decode_responses=True)

エビクションポリシー

ポリシー	動作	最適な用途
noeviction	満杯時に書き込みエラー	キュー / 重要なデータ
allkeys-lru	最も最近使用されていないものをエビクション	一般的なキャッシュ
volatile-lru	TTLのあるキーのみLRU	混合データストア
allkeys-lfu	最も頻繁に使用されていないものをエビクション	偏ったアクセスパターン
volatile-ttl	最も早く期限切れになるものをエビクション	長寿命データを優先

redis.conf で設定: maxmemory-policy allkeys-lru

アンチパターン

アンチパターン	問題点	修正方法
TTLなしのキー	メモリが無制限に増加	常にTTLを設定
本番環境での KEYS *	サーバーをブロック（O(N)）	SCAN カーソルを使用
大きなブロブの保存（>100KB）	シリアライズが遅い、メモリプレッシャー	参照を保存し、オブジェクトストアからフェッチ
すべてに単一のRedisを使用	キャッシュとキューの分離なし	別のDBまたはインスタンスを使用
コネクションプール制限の無視	負荷時の接続枯渇	プールサイズをワークロードに調整
キャッシュミスのスタンピード処理の欠如	コールドスタート時のスレッディングハード	ロックまたは確率的早期期限切れを使用
無計画な FLUSHALL	インスタンス全体をワイプ	キーパターンで削除範囲を限定

キャッシュミススタンピードの防止

import threading

_locks: dict[str, threading.Lock] = {}
_locks_mutex = threading.Lock()

def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = r.get(key)
    if cached:
        return json.loads(cached)

    with _locks_mutex:
        if key not in _locks:
            _locks[key] = threading.Lock()
        lock = _locks[key]
    with lock:
        cached = r.get(key)  # ロック獲得後にもう一度チェック
        if cached:
            return json.loads(cached)
        value = fetch_fn()
        r.setex(key, ttl, json.dumps(value))
        return value

注：マルチプロセスデプロイメントでは、インプロセスロックを上記「分散ロック」セクションの acquire_lock / release_lock に置き換えてください。

例

Django/Flask APIエンドポイントへのキャッシュ追加: キャッシュアサイドを setex と応答への5分間のTTLで使用します。リクエストパラメータをキーにします。

ユーザーごとのAPIレート制限: 低トラフィックエンドポイントには pipeline(transaction=True) を使用した固定ウィンドウを、正確なユーザーごとのスロットリングにはスライディングウィンドウLuaを使用します。

ワーカー間でのバックグラウンドジョブ調整: 予想されるジョブ実行時間を超えるTTLで acquire_lock を使用します。常に finally ブロックで解放します。

複数のサブスクライバーへの通知のファンアウト: ファイアアンドフォーゲットにはPub/Subを使用します。Lateコンシューマー向けの配信保証やリプレイが必要な場合はStreamsに切り替えます。

クイックリファレンス

パターン	使用場面
キャッシュアサイド	読み取りが多い、わずかな古さを許容
ライトスルーキャッシュ	強い一貫性が必要
分散ロック	リソースへの同時アクセスを防止
スライディングウィンドウレート制限	正確なユーザーごとのスロットリング
Redis Streams	コンシューマグループ付きの永続イベントキュー
Pub/Sub	配信保証不要のブロードキャスト
ソート済みセットリーダーボード	ランク付けスコアリング、ページネーション
HyperLogLog	低メモリでの近似ユニークカウント

関連

• スキル: postgres-patterns — リレーショナルデータパターン
• スキル: backend-patterns — APIおよびサービス層パターン
• スキル: database-migrations — スキーマバージョニング
• スキル: django-patterns — Djangoキャッシュフレームワーク統合
• エージェント: database-reviewer — 完全なデータベースレビューワークフロー