【実務・中級編】 List型：ブロッキング操作 – Redis

Redis List型ブロッキング操作の真髄：堅牢なメッセージングキュー設計のアンチパターンと極意

テックリードの私だ。コードレビューや設計レビューで、こんなコードを見かけて冷や汗をかいたことはないか？

良くある（そして本番で大惨事を引き起こす）ポーリング実装
while True:
item = redis_client.lpop(“my_queue”)
if item:
process(item)
else:
time.sleep(0.1) # CPUを無駄に燃やすな！

RedisのList型をただの「配列の延長」だと思っていないか？
`LPUSH` と `RPOP` を組み合わせたキュー構造は強力だが、素朴な実装をすれば、CPU使用率の不必要な高騰（ビジネスポーリング）、ネットワーク帯域の無駄遣い、そして最悪の場合のデータ消失というエンジニアの悪夢を直撃する。

今回は、RedisのList型におけるブロッキング操作（`BLPOP`, `BRPOP`, `BRPOPLPUSH` / `RPOPLPUSH`）に焦点を当て、プロダクション環境で耐えうる堅牢なメッセージングシステムの設計パターンを伝授する。

—

1. なぜブロッキング操作（Blocking Operations）が必要なのか？

先ほどのPythonコードの何がダメか？
データが入っていない間も、クライアントは無限に `LPOP` を叩き続ける（忙しい待機＝Busy Waiting）。これはRedisサーバーにもクライアントアプリにも無駄な負荷を強いる最悪のアンチパターンだ。

ここで登場するのが、`BLPOP` と `BRPOP` である。

これらは、「リストが空の場合に、データがプッシュされるまでコネクションをブロック（待機）する」操作だ。Redisのシングルスレッドモデルの特性を活かし、効率的にCPUを休止させつつ、データが投入された瞬間にミリ秒単位の遅延で処理を再開できる。

基本メカニズムのイメージ

[Client A: BLPOP queue] —> (データなし：ブロック中…)
|
[Client B: LPUSH queue “data”] ——->+ (データ投入！)
|
[Client A: BLPOP queue] <--- "data" を即座に取得して復帰 ---

2. コマンドの深掘り：BLPOP, BRPOP, そして安全な BRPOPLPUSH

実務でキューを設計する際、どのコマンドを選ぶべきか。それぞれの特性と「落とし穴」を理解してほしい。

`BLPOP` vs `BRPOP`

  • `BLPOP key [key …] timeout`: リストの左側（Head）からポップする。
  • `BRPOP key [key …] timeout`: リストの右側（Tail）からポップする。

一般的なFIFO（First-In-First-Out）キューを作る場合：

  • エンキュー（書き込み）：`LPUSH queue_name item`
  • デキュー（読み取り）：`BRPOP queue_name 0` （タイムアウト0は無限待機）

これで美しいFIFOキューが完成する。複数のキーを同時に指定して「優先度付きキュー」を作ることも可能だ（例: `BRPOP high_priority normal_priority 0`）。

—

最重要：『At-Least-Once（最低1回は配信）』を担保する BRPOPLPUSH

ここからが本題だ。`BRPOP` には致命的な弱点がある。
「ポップした瞬間にクライアントがクラッシュしたら、データは永遠に失われる（ロストする）」という点だ。

金融系や厳密なタスク処理基盤において、データロストは許されない。そこで使われるのが `BRPOPLPUSH`（Redis 6.2以降は `BLMOVE` の方が推奨されるが、概念は同じだ）である。

このコマンドは、以下の処理をアトミック（不可分）に実行する。
1. ソースキューから要素をポップする。
2. その要素を、同時に「処理中キュー（Processing Queue）」へプッシュする。
3. クライアントにその要素を返す。

[Source Queue] ──(BRPOPLPUSH)──> [Processing Queue (退避用)]
│
└──> クライアントへ返却

もしクライアントがメッセージ処理中に死亡しても、データは「処理中キュー」に残っているため、別のワーカーがリカバリできる。これが信頼性の高いメッセージングシステムの基礎だ。

—

3. 実践：Pythonによる堅牢な「Reliable Queue」の実装パターン

百聞は一見に如かず。プロダクションレベルで耐えうる、安全なワーカーのコードパターンを示そう。

import redis
import time
import json

class ReliableWorker:
def __init__(self, redis_client, source_queue, processing_queue):
self.r = redis_client
self.source = source_queue
self.processing = processing_queue

def run(self):
print(f”[] Worker started. Listening to {self.source}…”)
while True:
try:
# BRPOPLPUSHでアトミックにソースから処理中キューへ移動
# タイムアウトは5秒に設定し、定期的に生存確認できるようにする
item = self.r.brpoplpush(self.source, self.processing, timeout=5)

if not item:
continue # タイムアウト時はループ継続

self.process_message(item)

except redis.RedisError as e:
print(f”[!] Redis error occurred: {e}”)
time.sleep(2) # 障害時は少し待ってからリトライ
except Exception as e:
print(f”[!] Unexpected error: {e}”)
# ここでデッドレターキュー（DLQ）への退避ロジックなどを挟む

def process_message(self, raw_data):
data = json.loads(raw_data.decode(‘utf-8’))
print(f”[+] Processing task ID: {data.get(‘id’)}”)

# — 実際のビジネスロジック —
# time.sleep(1) # 処理のシミュレーション

# 処理が正常終了したら、処理中キューから削除（LREM）
# 引数: name, count, value
self.r.lrem(self.processing, 1, raw_data)
print(f”[x] Task completed: {data.get(‘id’)}”)

if __name__ == “__main__”:
client = redis.Redis(host=’localhost’, port=6379)
worker = ReliableWorker(client, “task_queue”, “task_queue:processing”)
worker.run()

この設計の美しさ（アーキテクチャのポイント）

1. アトミックな移動: メッセージが消える隙を与えない。
2. タイムアウトの活用: `timeout=5` を入れることで、ワーカーがグレースフルシャットダウン（安全な終了）シグナルを検知しやすくなる。無限ブロック（`0`）は極力避けろ。
3. 明示的なLREM: 処理が完全に成功したときだけ `LREM` で処理中キューから消し込む。

—

4. プロダクション運用の罠：パフォーマンスと運用の注意点

チーフアーキテクトとして、現場で絶対に踏んではならない地雷を共有しておく。

① 巨大なListを作るな（O(N)の呪縛）

RedisのList型は内部的には「双方向連結リスト（厳密にはQuicklistという最適化された構造）」だが、`LREM` や一部のインデックス操作は $O(N)$ の計算量を持つ。
数百万件の要素が詰まったListに対して `LREM` を実行すると、その瞬間Redisのメインスレッドがブロックされ、他のすべてのリクエストが数ミリ秒〜数百ミリ秒停止する（レイテンシスパイクの発生）。

  • 対策: キューのアイテムサイズは常に小さく保つ。数万件を超えたらアラートを上げ、コンシューマーのスケールアウトを検討せよ。

② コネクションタイムアウトの罠（TCP Timeout）

`BRPOP` や `BLPOP` で無限（`0`）にブロックしていると、途中のファイアウォールやロードバランサー（AWS ALBなど）が「アイドル状態のコネクション」とみなして勝手に切断することがある。

  • 対策: アプリケーション側で適切なタイムアウト（例: `30秒` 〜 `60秒`）を設定し、定期的にブロッキングを解除して生存確認（Ping/Pongなど）を行わせる設計にしろ。

③ クラッシュリカバリ（ゾンビデータの掃除）

先ほどの `BRPOPLPUSH` パターンにおいて、ワーカーが処理中に強制終了（OOM Killerやサーバー断）した場合、「処理中キュー」にデータが残ったままになる。

  • 対策: 定期実行バッチ（Cronなど）を用意し、処理中キューに長時間留まり続けている古いデータを検知し、元のソースキューに戻す（あるいはDead Letter Queueに送る）「リカバープロセス」を必ず実装すること。

—

5. まとめ

RedisのList型ブロッキング操作は、正しく使えば「軽量、高速、かつ堅牢なメッセージングシステム」の強力な武器となる。

  • 安易なポーリングは書くな。`BLPOP` / `BRPOP` を使え。
  • データの消失を防ぎたければ、`BRPOPLPUSH`（または `BLMOVE`）による2重キュー構造を採用せよ。
  • リストの肥大化とネットワークタイムアウトに気を配れ。

メッセージングの規模が大きくなり、Pub/SubやConsumer Group（Streams型）が必要になる手前のスケールであれば、このList型キューパターンで十分すぎるほどのパフォーマンスを発揮する。

設計レビューでこのパターンの実装を見かけたら、「お、わかっているね」と声をかけてやってほしい。健闘を祈る。

スポンサーリンク

コメント

タイトルとURLをコピーしました