redis-py活用ガイド:Caching、Pipeline、Distributed LockでPythonアプリのパフォーマンスを最適化する

Python tutorial - IT technology blog
Python tutorial - IT technology blog

背景:データベースがボトルネックになるとき

以前、よくある問題に遭遇したことがある。Pythonスクリプトで100Kレコードを処理し、各レコードに設定情報を取得するためのデータベースクエリが必要だった。結果は?スクリプトの実行に20分以上かかり、データベースのCPUが90%まで跳ね上がった。問題は処理ロジックではなく、同じクエリが何千回も繰り返し呼び出され、毎回同じ結果を返していた点にある。

Redisとredis-pyがそのスクリプトを救ってくれた。データを読み取るたびにデータベースを呼び起こすのではなく、Redisが結果をメモリに保持する——読み書き速度はミリ秒ではなくマイクロ秒単位だ。cacheレイヤーを追加するだけで、20分かかっていたスクリプトが90秒まで短縮された。アルゴリズムを書き直したわけではなく、ただ同じ処理を繰り返すのを避けただけだ。

本記事では、redis-pyで最もよく使う3つの具体的なテクニックに焦点を当てる:Caching(無駄なクエリを回避)、Pipeline(コマンドをbatchにまとめてネットワークのround-tripを削減)、Distributed Lock(複数のprocessが並列実行する際の一貫性を保証)。

redis-pyのインストール

まずRedis serverが起動している必要がある。最も手軽な方法はDockerを使うことだ:

docker run -d --name redis-local -p 6379:6379 redis:7-alpine

次に、Pythonライブラリをインストールする:

pip install redis
# hiredisを追加してレスポンスのパースを高速化する
pip install redis[hiredis]

hiredisはレスポンスのパースを高速化するC extensionであり、使わない理由はない。接続を確認する:

import redis

r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
print(r.ping())  # 接続成功の場合はTrue

decode_responses=Trueに注意——このパラメータがないと、返り値がすべてbytesになる。使い始めた頃はよくこれを忘れて、string comparisonがなぜ動かないのか10分間デバッグしたものだ。

詳細な設定

1. Cache-Aside PatternによるCaching

考え方はシンプルだ:まずRedisから読み取り、missの場合のみDBに問い合わせる——そしてTTL付きで結果をRedisに保存し、次回はDBへのアクセスを避ける。

import redis
import json

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

def get_user_config(user_id: int) -> dict:
    cache_key = f"user:config:{user_id}"

    # まずcacheから読み取る
    cached = r.get(cache_key)
    if cached:
        return json.loads(cached)

    # cache miss — DBから読み取る
    config = fetch_from_database(user_id)  # 実際のDBクエリ関数

    # cacheに保存、TTL 5分
    r.setex(cache_key, 300, json.dumps(config))
    return config

100Kレコードのうち固有のuser_idが500件しかない場合、DBクエリ数は100Kから約500に減る——200分の1だ。データベースCPUは90%から約15%に低下し、スクリプトは20分から90秒になる。数十行のコードを追加するだけで、リファクタリングは不要だ。

各関数にboilerplateを書きたくない場合は、decoratorでラップできる:

from functools import wraps

def redis_cache(ttl=300):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            key = f"cache:{func.__name__}:{args}:{sorted(kwargs.items())}"
            result = r.get(key)
            if result:
                return json.loads(result)
            result = func(*args, **kwargs)
            r.setex(key, ttl, json.dumps(result))
            return result
        return wrapper
    return decorator

@redis_cache(ttl=600)
def get_product_details(product_id: int) -> dict:
    # Query database...
    pass

2. Pipeline — ネットワークのRound-tripを削減

Redisコマンドを1つ送るたびにネットワークのround-tripが発生する。latencyが1msのネットワークで1000個のコマンドを個別に実行すると、ネットワークのoverheadだけで1秒かかる。Pipelineは複数のコマンドを1つのbatchにまとめ、一度に送信して一度に結果を受け取る。

import redis
import time

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

# pipelineを使わない方法(遅い)
def set_scores_naive(scores: dict):
    start = time.perf_counter()
    for user_id, score in scores.items():
        r.set(f"score:{user_id}", score)
    print(f"Naive: {time.perf_counter() - start:.3f}s")

# Pipelineを使う方法(8〜12倍速い)
def set_scores_pipeline(scores: dict):
    start = time.perf_counter()
    with r.pipeline(transaction=False) as pipe:  # transaction=FalseでスループットがUpする
        for user_id, score in scores.items():
            pipe.set(f"score:{user_id}", score)
        pipe.execute()  # まとめて一度に送信
    print(f"Pipeline: {time.perf_counter() - start:.3f}s")

# 10,000レコードでテスト
test_scores = {f"user_{i}": i * 10 for i in range(10000)}
set_scores_naive(test_scores)    # localhostで約2.5s
set_scores_pipeline(test_scores) # localhostで約0.3s

Pipelineは複数のkeyを同時に読み取る場合にも有効だ:

def get_multiple_users(user_ids: list) -> dict:
    with r.pipeline(transaction=False) as pipe:
        for uid in user_ids:
            pipe.get(f"user:config:{uid}")
        results = pipe.execute()

    return {
        uid: json.loads(res) if res else None
        for uid, res in zip(user_ids, results)
    }

transaction=FalseはMULTI/EXECのwrappingを無効にしてスループットを向上させる。読み取った値に基づいて書き込む場合など、本当にatomic executionが必要な場合にのみtransaction=Trueを有効にする。

3. Distributed Lock — Race Conditionを防ぐ

実際のシナリオ:3つのworker processが「メールを送信済みか?」を同時にチェックし、いずれも「まだ、自分が送る」と判断する。結果として、ユーザーは同じメールを3通受け取る。Distributed lockは、コードの特定の部分を一度に1つのprocessだけが実行できるようにすることでこれを防ぐ。

import redis

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

def send_notification_once(user_id: int, message: str) -> bool:
    lock_key = f"lock:notification:{user_id}"

    # timeout 10秒でlockの取得を試みる
    lock = r.lock(lock_key, timeout=10, blocking_timeout=5)

    acquired = lock.acquire(blocking=True)
    if not acquired:
        print(f"user {user_id}のlockを取得できませんでした。スキップします")
        return False

    try:
        # lock取得後にdouble-check
        sent_key = f"notification:sent:{user_id}"
        if r.exists(sent_key):
            return False

        # notificationを送信
        send_email(user_id, message)

        # 送信済みとしてマーク、TTL 1時間
        r.setex(sent_key, 3600, "1")
        return True

    finally:
        lock.release()  # finallyで必ずreleaseする

lock.release()は必ずfinallyブロック内に記述する——例外はない。releaseせずに例外が発生すると、timeoutが自動的に期限切れになるまで他のすべてのprocessがブロックされる。覚えておくべき2つのパラメータ:timeoutはprocessがクラッシュした場合にlockが自動expireするまでの時間(safety net)、blocking_timeoutはprocessがlockを取得するまで待機できる最大時間だ。

もう1つよくあるケースは、cron jobの並列実行だ:

def run_scheduled_job(job_name: str):
    lock_key = f"job:running:{job_name}"
    lock = r.lock(lock_key, timeout=300)  # Jobの最大実行時間は5分

    if not lock.acquire(blocking=False):  # Non-blocking: 実行中の場合はskip
        print(f"Job {job_name}は別のworkerで実行中です。スキップします")
        return

    try:
        execute_job(job_name)
    finally:
        lock.release()

テスト & Monitoring

Cache hit rate

Redisはメモリにhitとmissをトラッキングしているためexternal toolは不要だ。以下のコマンドで即座に数値を確認できる:

redis-cli info stats | grep -E "keyspace_hits|keyspace_misses"
# 出力例:
# keyspace_hits:45231
# keyspace_misses:1823
# Hit rate = 45231 / (45231 + 1823) ≈ 96%
def get_cache_stats() -> dict:
    info = r.info('stats')
    hits = info['keyspace_hits']
    misses = info['keyspace_misses']
    total = hits + misses
    hit_rate = (hits / total * 100) if total > 0 else 0

    return {
        'hits': hits,
        'misses': misses,
        'hit_rate': f"{hit_rate:.1f}%",
    }

def get_memory_usage() -> dict:
    info = r.info('memory')
    return {
        'used': info['used_memory_human'],
        'fragmentation_ratio': info['mem_fragmentation_ratio'],
    }

Slow logとmemory設定

# slow logを有効化 — 1ms以上かかるコマンドをログに記録
redis-cli config set slowlog-log-slower-than 1000  # マイクロ秒単位
redis-cli slowlog get 10

# memoryを制限し、LRUで古いkeyを自動evict
redis-cli config set maxmemory 512mb
redis-cli config set maxmemory-policy allkeys-lru

allkeys-lruポリシーはほとんどのcachingユースケースに適している:メモリが満杯になった際、OOMエラーを返す代わりにRedisが最も使われていないkeyを自動的にevictする。maxmemoryの設定はproductionに展開する前に必ず行うべきステップだ。

Pipelineの実際のBenchmark

import time

def benchmark_pipeline(n=5000):
    test_data = {f"key:{i}": f"value:{i}" for i in range(n)}

    start = time.perf_counter()
    for k, v in test_data.items():
        r.set(k, v)
    t_normal = time.perf_counter() - start

    start = time.perf_counter()
    with r.pipeline(transaction=False) as pipe:
        for k, v in test_data.items():
            pipe.set(k, v)
        pipe.execute()
    t_pipeline = time.perf_counter() - start

    print(f"Normal:   {t_normal:.3f}s")
    print(f"Pipeline: {t_pipeline:.3f}s")
    print(f"Speedup:  {t_normal / t_pipeline:.1f}x")

benchmark_pipeline()

localhost環境(Redis 7、Python 3.11)では、5000 operationsで8〜12倍高速になる。しかし、latencyが10〜50msの実際のネットワーク経由で接続する場合、差はさらに大きくなる——1000コマンド × 10ms = 10秒のoverhead。pipelineを使えば、1000コマンドすべてを1回のround-tripで処理できる。

この3つのテクニックは独立しており、直面している問題に応じて使い分ける。データベースCPUが異常に高い場合はCaching。大量の読み書きが必要な場合はPipeline。複数のworkerが同じデータストリームを処理する場合はDistributed lock。適切な場所に適用すれば、redis-pyはほとんど期待を裏切らない。

Share: