背景：データベースがボトルネックになるとき
以前、よくある問題に遭遇したことがある。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()) # 接続成功の場合はTruedecode_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 config100Kレコードのうち固有の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...
pass2. 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.3sPipelineは複数の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-lruallkeys-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はほとんど期待を裏切らない。
