Redis メッセージキューのメモ：List と Pub/Sub

一、メッセージキューを実装する理由

メッセージキューの主な役割は、「メッセージの生成」と「メッセージの処理」を分離することです。

例えば、注文業務の場合：

ユーザーが注文を行う
    ↓
プロデューサーが注文タスクを Redis に格納する
    ↓
コンシューマーが Redis から注文タスクを取り出す
    ↓
非同期でデータベースに注文を登録する

これにより、以下のことが実現できます：

ピーク負荷の平準化：高並行性のリクエストはまず Redis に入り、データベースは徐々に処理します。
非同期処理：ユーザーはすべての業務処理が完了するのを待つ必要がありません。
疎結合：プロデューサーはタスクの送信のみを担当し、コンシューマーはタスクの処理のみを担当します。
マルチインスタンス共有：複数のアプリケーションが同一の Redis キューを利用可能。

Redis で一般的なメッセージ通信方式には以下がある：

List：競合型消費。単純なタスクキューに適している。
Pub/Sub：ブロードキャスト型消費。リアルタイム通知に適している。
Stream：信頼性が高く、注文やタイムセールなどの重要な業務に適しています。

二、Redis List によるメッセージキューの実装

1. List の基本概念

Redis List は順序付きの両端リストであり、左側または右側から要素の追加・取得が可能です。

最も一般的なキューの構成：

プロデューサー：RPUSH
コンシューマー：BLPOP

ロジックは以下の通りです：

左側　右側
← BLPOP [メッセージ1、メッセージ2、メッセージ3] RPUSH →

プロデューサーは右側からメッセージを追加し、コンシューマーは左側からメッセージを取得するため、先入れ先出し（FIFO）が満たされます。

先に投入されたメッセージは、通常、先に処理されます。

これが FIFO です：

First In First Out、先入れ先出し。

2. List の常用コマンド

コマンド	機能
`LPUSH key value`	左側から要素を追加
`RPUSH key value`	右側から要素を追加
`LPOP key`	左側から要素を取り出して削除
`RPOP key`	右側から要素を取り出して削除する
`BLPOP key timeout`	左側からブロックして要素を取得する
`BRPOP key timeout`	右側からブロックして要素を取得する
`LRANGE key 0 -1`	リストの全内容を閲覧
`DEL key`	リストを削除

3. 最も単純なリストキュー

注文キューの名前を以下のように仮定する：

order.queue

プロデューサーが注文メッセージを送信：

RPUSH order.queue 「order-1001」
RPUSH order.queue 「order-1002」
RPUSH order.queue 「order-1003」

このときの論理構造は次の通りです：

左側　右側
[ order-1001 , order-1002 , order-1003 ]

すべてのメッセージを確認：

LRANGE order.queue 0 -1

結果は次のようになります：

1) 「order-1001」
2) 「order-1002」
3) 「order-1003」

コンシューマーがメッセージを1つ取得：

LPOP order.queue

結果：

「order-1001」

この時点でキューは次のようになります：

[ order-1002 , order-1003 ]

注意：

LPOPは単にメッセージを読み取るだけでなく、メッセージを取り出して削除します。

4. なぜ実際の消費では通常 BLPOP

が使用されるのか。キューが空の場合：

LPOP order.queue

は以下を返す：

(nil)

もしコンシューマーが LPOP を繰り返し実行し続けると：

LPOP
メッセージなし
LPOP
メッセージなし
LPOP
メッセージなし

これはポーリングと呼ばれ、Redis に繰り返しアクセスするため、CPU および Redis のリソースを浪費します。

そのため、実際のコンシューマーでは通常、ブロッキングコマンドが使用されます：

BLPOP order.queue 0

意味：

キューにメッセージがある場合、直ちに1つ取り出します。

キューが空の場合、現在のクライアントはブロックされて待機します。

0はタイムアウトなしを意味し、ずっと待機し続けます。

これはJavaの以下のコードに似ています：

orderTasks.take();

対応関係：

Java BlockingQueue	Redis List
`put(message)`	`RPUSH queue message`
`take()`	`BLPOP queue 0`
`poll()`	`LPOP queue`

5. 複数の消費者による競合消費

複数の端末を開き、それぞれを異なる消費者として扱うことができます。

端末 A：プロデューサー
端末 B：消費者 c1
端末 C：消費者 c2

消費者 c1：

BLPOP order.queue 0

消費者 c2：

BLPOP order.queue 0

プロデューサーがメッセージを送信：

RPUSH order.queue 「order-1001」

Redisはこのメッセージを、待機中のいずれかのコンシューマーに渡します。

c1 または c2 のうち、どちらか一方だけが order-1001 を受け取ることができます。

もう一方のコンシューマーは、同じメッセージを受け取ることはありません。

これを

競合消費。

複数のコンシューマーがキュー内のメッセージを競合して取得することで、処理能力を向上させます。

注意：

BLPOP は一度に 1 つのメッセージしか取得しません。

メッセージを取得すると、コマンドは終了します。

消費を継続するには、再度 BLPOP を実行する必要があります。

将来的には Java にループが導入される予定です：

while (true) {
    // メッセージの取得をブロック
    // メッセージを処理
}

6. List が存在しない場合、自動的に作成されるか

プロデューサーが初めて実行する場合：

RPUSH order.queue 「order-1001」

もし order.queue が存在しない場合、Redis は自動的に List を作成し、そこにメッセージを書き込みます。

したがって、事前にキューを作成する必要はありません。

各コマンドの動作：

は作成されない。

操作	キューが存在しない場合の動作
`RPUSH` / `LPUSH`	自動的に List を作成し、メッセージを書き込み
`LPOP` / `RPOP`	`(nil)`を返す。
`BLPOP` / `BRPOP`	ブロックして待機する。
最後のメッセージを取り出した後	空のリストは通常自動的に削除される

例えば、コンシューマーがまず以下を実行する場合：

BLPOP order.queue 0

たとえ order.queue が存在しなくても、ブロックして待機します。

その後、プロデューサーが以下を実行します：

RPUSH order.queue 「order-1001」

Redisは自動的にListを作成し、待機中のコンシューマーに直ちにメッセージを渡しします。

7. Listの信頼性に関する問題

通常のListキューの消費プロセス：

キューにメッセージがある
    ↓
コンシューマーによるBLPOP / LPOP
    ↓
メッセージは直ちにListから削除される
    
↓
コンシューマーが処理を開始

もしコンシューマーがメッセージを取得した直後にクラッシュした場合：

コンシューマーが order-1001 を取得
    ↓
Redis はすでに order-1001 を削除済み
    ↓
コンシューマーが MySQL に書き込む前にクラッシュ
    ↓
メッセージが失われる

したがって：

RPUSH + BLPOP はシンプルですが、メッセージが確実に処理完了することを保証することはできません。

適している用途：

ログ、単純な通知、少量の損失が許容される非同期タスク。

適していない用途：

注文の作成、決済、在庫の減算、タイムセール注文などの重要な業務。

8. Listの改良：処理中キュー

2つのListを用意できます：

order.queue 処理待ちキュー
order.processing 処理中キュー

プロデューサーがメッセージを送信：

RPUSH order.queue 「order-1001」

コンシューマーはメッセージを直接削除せず、以下のようにします：

BRPOPLPUSH order.queue order.processing 0

意味：

order.queueの右側からメッセージを1つ取り出し、
order.processingの左側に配置します。
この移動プロセス全体はアトミックです。

フロー：

処理待ちキュー
    ↓
アトミックな移動
    ↓
処理中キュー
    ↓
ビジネス処理成功
    ↓
処理中キューからメッセージを削除

処理成功後：

LREM order.processing 1 「order-1001」

コンシューマーがダウンした場合：

メッセージは order.processing に残ったままとなり、
すぐに消えることはありません。

ただし、この方式では、リトライ、タイムアウトチェック、重複消費の制御、失敗記録などのロジックをすべて独自に実装する必要があります。

したがって、Listは信頼性を向上させることができますが、実装の複雑さは徐々にStreamに近づいていきます。

三、Redis Pub/Sub パブリッシュ・サブスクライブ

1. Pub/Subの基本概念

Pub/SubはPublish / Subscribeの略語であり、つまり「配信／購読」を意味します。

これは、従来の「1人の消費者が1つのメッセージを取得する」というキューではなく、ブロードキャストの仕組みです。

パブリッシャー (Publisher)
    ↓ PUBLISH
チャンネル (Channel)
    ↓ SUBSCRIBE
複数のサブスクライバー (Subscriber)

例：

パブリッシャー
    ↓
news
    ├── サブスクライバー A が受信
    ├── サブスクライバー B が受信
    └── サブスクライバー C が受信

1つのメッセージは、そのチャンネルを購読している、現在オンライン中のすべてのクライアントに受信されます。

2. Pub/Sub の主要コマンド

コマンド	機能
`PUBLISH channel message`	チャンネルへのメッセージ配信
`SUBSCRIBE channel`	チャンネルを厳密に購読する
`UNSUBSCRIBE channel`	厳密な購読を解除する
`PSUBSCRIBE pattern`	パターンに基づいてチャンネルを購読する
`PUNSUBSCRIBE pattern`	パターンによる購読を解除

3. 基本的な配信と購読

端末 A が購読者の場合：

SUBSCRIBE news

チャンネルを購読します：

news

端末 B がパブリッシャーとして：

PUBLISH news 「hello redis」

返される値：

(整数) 1

現在、1人のサブスクライバーがメッセージを受信したことを示します。

端末Aは以下を受信します：

1) 「message」
2) 『news』
3) 「hello redis」

意味：

message：通常のチャンネルメッセージを受信したことを示す
news：メッセージが属するチャンネル
hello redis：メッセージの内容

4. Pub/Subのブロードキャスト機能

2人のサブスクライバーが以下を実行したと仮定する：

SUBSCRIBE news

その後、パブリッシャーが以下を送信します：

PUBLISH news 「タイムセール開始」

2人のサブスクライバーは以下を受信します：

タイムセール開始

これはListとは異なります。

List：

1つのメッセージは通常、1人のコンシューマーによってのみ取得されます。

Pub/Sub：

1つのメッセージは、オンラインのすべてのサブスクライバーにブロードキャストされます。

5. Pub/Subが注文メッセージに適さない理由

Pub/Subは過去のメッセージを保存しません。

購読者がいない状態でパブリッシュした場合：

PUBLISH news 「イベント開始」

返される値：

(整数) 0

これは、現在オンラインの購読者がいないことを示しています。

このメッセージは保存されません。

その後、誰かが次のように実行しても：

SUBSCRIBE news

先のメッセージは受信できません。

したがって、Pub/Subの特徴は：

まずサブスクライブして初めて、その後配信されるメッセージを受信できることです。

サブスクライバーが切断されたり、システムがダウンしたり、ネットワークに異常が発生した場合、切断期間中のメッセージは直接見逃されてしまいます。

Pub/Subには以下の機能がありません：

メッセージの永続化
確認メカニズム
Pending List
失敗時の再試行
コンシューマーグループ

したがって、以下には適していません：

注文
決済
在庫の減算
タイムセールでの注文

これらの業務では、メッセージの確実な処理が保証されなければなりません。

6. Pub/Sub が適しているシナリオ

Pub/Sub は、「すべてのオンラインクライアントが即座に把握すべき」通知型のシナリオに適しています。

例：

チャットルームでのメッセージ配信
システムメンテナンスの通知
キャッシュの失効通知
オンラインユーザーのステータス変化
リアルタイム監視ダッシュボードの更新
WebSocket プッシュによる補助通知

Pub/Sub は次のように理解できます：

WeChat グループの告知。

オンラインの人はすぐに確認できますが、オフラインの人は見逃してしまいます。

7. 複数のチャンネルの購読

一度に複数のチャンネルを購読できます：

SUBSCRIBE news system.notice cache.clear

これにより、いずれかのチャンネルにメッセージが届くと、現在のクライアントはすべて受信できるようになります。

例：

PUBLISH news 「ニュース更新」
PUBLISH system.notice 『10分後にサーバーメンテナンスを実施します』
PUBLISH cache.clear 「商品キャッシュをクリアしてください」

8. 通常サブスクリプションの解除

特定のチャンネルを解除するには：

UNSUBSCRIBE news

現在のクライアントのすべての通常サブスクリプションを解除するには：

UNSUBSCRIBE

対応関係：

SUBSCRIBE 特定のチャンネルを購読
UNSUBSCRIBE 特定のチャンネルの購読を解除

9. パターン購読：PSUBSCRIBE

特定のチャンネルを購読するほか、パターンに基づいて購読することも可能です。

例：

PSUBSCRIBE news.*

は、news.で始まるすべてのチャンネルを購読することを意味します。

は以下と一致します：

news.sports
news.tech
news.game
news.local

パブリッシャーが実行：

PUBLISH news.sports 「試合開始」

あるいは：

PUBLISH news.tech 「Redis 通知」

パターン購読者は全員、このメッセージを受信します。

パターン購読で受信される形式は次のようなものです：

1) 「pmessage」
2) 「news.*」
3) 「news.sports」
4) 「試合開始」

意味：

pmessage はパターンサブスクリプションメッセージを表します
news.* は現在のサブスクリプションパターン
news.sports は実際に公開されたチャンネル
試合開始 はメッセージの内容

10. PUNSUBSCRIBE の役割

PUNSUBSCRIBE は、パターンサブスクリプションを解除するコマンドです。

例えば、以前に以下を実行した場合：

PSUBSCRIBE news.*

これを解除するには：

PUNSUBSCRIBE news.*

すべてのモードの購読を解除：

PUNSUBSCRIBE

対応関係：

登録方法	解除方法
`SUBSCRIBE news`	`UNSUBSCRIBE news`
`PSUBSCRIBE news.*`	`PUNSUBSCRIBE news.*`

したがって：

PUNSUBSCRIBE の意味は Pattern Unsubscribe、
すなわちパターン登録の解除です。

11. 通常の登録とパターン登録の重複の問題

同じクライアントが、

SUBSCRIBE news.sports

さらに以下を実行するとします：

PSUBSCRIBE news.*

その後、以下をパブリッシュします：

PUBLISH news.sports 「試合開始」

このクライアントは、2回のメッセージを受信する可能性があります：

1回はニュース.sportsへの精密サブスクリプションから
1回はニュース.*へのパターンサブスクリプションから

したがって、開発時には同一のビジネスロジックで同一チャンネルに対して重複したサブスクリプションを設定しないようにすべきです。そうしないと、通知が重複して処理される可能性があります。

4. List と Pub/Sub の比較

比較項目	Redis List	Redis Pub/Sub
コアパターン	タスクキュー	メッセージブロードキャスト
メッセージを処理する主体	1人のコンシューマー	すべてのオンラインサブスクライバー
メッセージの保存の有無	Listに一時的に保存	保存しない
コンシューマーがオフラインの場合	メッセージはキューに一時保存される	メッセージを直接見逃す
ブロック待機をサポートするか	サポート `BLPOP` / `BRPOP`	購読後は継続して受信
確認メカニズムがあるか	なし	なし
メッセージの損失が発生しやすいか	消費後にシステムがダウンすると損失の可能性がある	オフライン時や異常時に損失の可能性がある
代表的な用途	単純な非同期タスク	リアルタイムのブロードキャスト通知
注文処理に適しているか	信頼性が不十分	不向き

五、最終的な記憶

List キュー

最も一般的な記述法：

RPUSH order.queue 「メッセージ」
BLPOP order.queue 0

意味：

プロデューサーは右側からメッセージを投入し、
コンシューマーは左側からブロックしてメッセージを取得し、
先入れ先出し（FIFO）の競合消費を実現する。

長所：

シンプル
ブロックをサポート
複数のコンシューマーによる競合消費が可能

短所：

メッセージは取得後すぐに削除される
コンシューマーのダウンによりメッセージが失われる可能性がある
確認、再試行、コンシューマーグループの仕組みがない

Pub/Sub

最も一般的な記述法：

SUBSCRIBE news
PUBLISH news 「メッセージ」

意味：

パブリッシャーがチャンネルにメッセージをブロードキャストすると、
現在オンラインのすべてのサブスクライバーがそれを受信する。

サブスクリプション方式：

PSUBSCRIBE news.*
PUNSUBSCRIBE news.*

長所：

リアルタイム配信
使い方が簡単
通知系のシナリオに適している

デメリット：

メッセージが保存されない
サブスクライバーがオフラインになるとメッセージが失われる
確認や再試行がない
注文などの重要な業務には適さない

一言で区別すると

List：先着順で処理される。

Pub/Sub：購読した者は全員受信する。

Stream：メッセージの確実な割り当て、確認、復旧が行われる。