一、プロデューサーがメッセージを送信したことをどう保証するか？
まず重要な点:
プロデューサーが保証できるのは、メッセージが RocketMQ Broker に正常に送信されたことまでです。
プロデューサーは、コンシューマーが業務処理を完了したことまでは直接保証できません。
つまり RocketMQ は非同期モデルです:
Producer -> Broker -> Consumer
プロデューサーはメッセージ送信後、通常は次だけを知ることができます:
メッセージが Broker に正常に書き込まれたか
次のことではありません:
コンシューマーが最終的に処理を終えたか
以前の概念では、Broker が実際にメッセージを保存する場所であり、Producer は Broker に送信し、Consumer は Broker からメッセージを取得して処理します。
1. プロデューサーは「メッセージを送信した」とどう確認するか？
RocketMQ のメッセージ送信には通常 3 つの方法があります:
同期送信
非同期送信
一方向送信
1) 同期送信: 最も一般的で信頼性が高い
プロデューサーはメッセージ送信後、RocketMQ の結果を待ちます。
SendResult sendResult = rocketMQTemplate.syncSend(
"order_topic:create",
"注文作成成功、注文ID: 10001"
);
System.out.println(sendResult.getSendStatus());
もし次が返れば:
SEND_OK
メッセージは Broker に正常に送信されたことを意味します。
例えるなら:
宅配便を配送センターに渡し、配送センターから受領証を受け取った状態です。
2) 非同期送信: 高並列に向く
プロデューサーは送信後に現在のスレッドをブロックせず、RocketMQ が後でコールバックで成功か失敗を通知します。
rocketMQTemplate.asyncSend(
"order_topic:create",
"注文作成成功、注文ID: 10001",
new SendCallback() {
@Override
public void onSuccess(SendResult result) {
System.out.println("送信成功: " + result);
}
@Override
public void onException(Throwable cause) {
System.out.println("送信失敗: " + cause.getMessage());
}
}
);
3) 一方向送信: 最速だが信頼性はない
rocketMQTemplate.sendOneWay(
"access_log_topic",
"ユーザーアクセスログ"
);
結果を待たず、成功失敗も気にしません。
ログや計測のように多少の欠落が許される場面に向きます。
注文、決済、クーポンなど重要な業務には向きません。
二、コンシューマーがメッセージを完全に消費したことをどう保証するか？
ここは非常に注意が必要です:
RocketMQ 自体が保証できるのは、「コンシューマーが正常に戻ったらメッセージ消費成功とみなす」ところまでです。
あなたの業務が本当に成功したかは、コード設計に依存します。
コンシューマーのコードはおおよそ次のようになります:
@Component
@RocketMQMessageListener(
topic = "order_topic",
consumerGroup = "inventory_consumer_group"
)
public class InventoryConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String payload) {
// 1. 業務ロジックを実行する。例: 在庫を減らす
System.out.println("在庫を減らす: " + payload);
// 2. ここで例外を投げなければ、RocketMQ は消費成功とみなす
}
}
onMessage() が正常終了すれば、RocketMQ は次とみなします:
このメッセージは消費成功
例外を投げた場合:
@Override
public void onMessage(String payload) {
throw new RuntimeException("在庫減算に失敗");
}
RocketMQ は次とみなします:
このメッセージは消費失敗。再試行が必要
三、しかし「完全消費」とは何を意味するのか？
この言葉は分解して考える必要があります。
注文サービスが 1 件のメッセージを送ったとします:
ユーザーが注文に成功
そのとき 3 つのコンシューマーグループがあるとします:
inventory_group: 在庫を減らす
sms_group: SMS を送る
points_group: ポイントを加算する
すると「完全消費」には 2 通りの意味がありえます。
ケース 1: 特定のコンシューマーグループの消費成功だけを気にする
たとえば在庫システムが在庫減算に成功したかだけを気にする場合です。
その場合は次を見ます:
inventory_group が消費成功したか
在庫コンシューマーが正常終了し例外を投げなければ、RocketMQ はそのコンシューマーグループの消費成功とみなします。
ケース 2: すべての業務システムの消費成功を気にする
たとえば次を知りたい場合です:
在庫を引いた
SMS を送った
ポイントを加算した
これは RocketMQ が自動でやってくれることではありません。
自分で業務状態テーブルを設計する必要があります。
例えば:
CREATE TABLE order_message_consume_status (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
order_id BIGINT,
message_key VARCHAR(100),
inventory_status TINYINT,
sms_status TINYINT,
points_status TINYINT,
create_time DATETIME,
update_time DATETIME
);
各コンシューマーは処理後に自分の状態を更新します:
在庫システム消費成功 -> inventory_status = 1
SMS システム消費成功 -> sms_status = 1
ポイントシステム消費成功 -> points_status = 1
最後に次を判定します:
inventory_status = 1
sms_status = 1
points_status = 1
そうして初めて言えます:
この注文メッセージに関連する業務はすべて処理完了した。
四、RocketMQ 自体は「強い同期確認」のためのものではない
MQ を次のようなモデルと理解してはいけません:
プロデューサーがメッセージを送信
すべてのコンシューマーの処理完了を待つ
その後プロデューサーが処理を続ける
これでは MQ の意味がなくなります。
MQ の核心的価値は:
非同期
疎結合
ピークカット
最終整合性
したがって通常の設計は:
プロデューサーはメッセージが Broker に確実に届くことだけを保証する
コンシューマーは自分が最終的に処理成功できることを保証する
業務は状態テーブル、再試行、冪等性で最終整合性を保証する
つまり:
MQ システムでは通常「即時整合性」ではなく「最終整合性」を目指します。
五、全体の経路をより信頼できるものにするには？
一般的には次のことを行います。
1. プロデューサーは同期送信を使い、SEND_OK を確認する
SendResult result = rocketMQTemplate.syncSend(
"coupon_topic:expire",
couponMessage
);
if (result.getSendStatus() != SendStatus.SEND_OK) {
throw new RuntimeException("クーポン期限切れメッセージの送信に失敗");
}
これにより、プロデューサーはメッセージが Broker に正常に渡ったか分かります。
2. 重要な業務ではトランザクションメッセージを使う
例えば:
注文作成に成功したら
必ず注文作成メッセージを送る
先に注文を INSERT し、その後メッセージを送ると、次のようなことが起こりえます:
注文 INSERT 成功
メッセージ送信失敗
これでは不整合になります。
RocketMQ のトランザクションメッセージは、この問題を解決するためのものです。
おおよその流れは:
1. まずハーフメッセージを送信する
2. ローカルトランザクションを実行する。例: 注文を作成する
3. ローカルトランザクション成功ならメッセージをコミットする
4. ローカルトランザクション失敗ならメッセージをロールバックする
次のように理解できます:
注文作成に成功したときだけ、メッセージがコンシューマーから見える。
注文作成に失敗したら、メッセージはコンシューマーから見えない。
3. コンシューマーは処理成功時だけ正常に戻す
コンシューマー内で安易に例外を握りつぶしてはいけません。
間違った書き方:
@Override
public void onMessage(String payload) {
try {
// 在庫を減らす
inventoryService.deduct(payload);
} catch (Exception e) {
// ログだけ出して例外を投げない
e.printStackTrace();
}
}
この書き方は危険です。
在庫減算に失敗していても、メソッドは最終的に正常終了しているからです。
RocketMQ は次とみなします:
消費成功
正しい書き方:
@Override
public void onMessage(String payload) {
try {
inventoryService.deduct(payload);
} catch (Exception e) {
throw new RuntimeException("在庫減算に失敗。RocketMQ の再試行を待つ", e);
}
}
これで RocketMQ は再試行します。
4. コンシューマーは必ず冪等にする
RocketMQ はメッセージを重複配信する可能性があるためです。
たとえば同じ注文メッセージを 2 回消費した場合:
注文10001の在庫を減らす
注文10001の在庫を減らす
冪等にしないと、在庫を余分に減らしてしまいます。
データベースの一意インデックスで重複を防げます。
例えば:
CREATE TABLE consume_record (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
message_key VARCHAR(100) UNIQUE,
consumer_group VARCHAR(100),
create_time DATETIME
);
消費前に判定します:
@Override
public void onMessage(OrderMessage message) {
// 1. すでに消費済みか判定する
boolean alreadyConsumed = consumeRecordService.exists(message.getMessageKey());
if (alreadyConsumed) {
return;
}
// 2. 業務ロジックを実行する
inventoryService.deduct(message.getOrderId());
// 3. 消費済みとして記録する
consumeRecordService.save(message.getMessageKey(), "inventory_group");
}
これならメッセージが重複して来ても、在庫を重複して減らしません。
六、死信キューとは何か？
死信キューは英語で:
Dead Letter Queue
DLQ
意味は:
あるメッセージの消費が繰り返し失敗し、最大再試行回数を超えると、通常の配信は続けられず、死信キューに入ります。
例えば:
注文メッセージ -> 在庫コンシューマー
在庫コンシューマーがずっと失敗する場合:
1回目失敗
2回目失敗
3回目失敗
...
16回目失敗
再試行回数を超えると、このメッセージは死信キューに入ります。
七、なぜ死信キューが必要なのか？
処理できないままのメッセージがあるからです。
例えば:
注文IDが存在しない
データベースのカラム異常
業務データ形式の誤り
在庫システムのコードにバグ
外部インターフェースがずっと失敗
RocketMQ が無限に再試行し続けると、リソースを浪費し、他のメッセージを詰まらせる可能性があります。
そこで RocketMQ は次のように判断します:
このメッセージは何度も再試行したが、まだ失敗している。
ひとまず死信キューに入れて、後で人が調査できるようにしよう。
八、死信キューはどう理解すればよいか？
次のように理解できます:
問題のある荷物の一時保管場所
正常なメッセージフロー:
メッセージ -> コンシューマー -> 消費成功
失敗と再試行のフロー:
メッセージ -> コンシューマー -> 失敗
メッセージ -> コンシューマー -> 失敗
メッセージ -> コンシューマー -> 失敗
最終的に:
メッセージ -> 死信キュー
つまり:
正常キューで処理できない異常メッセージは、最後に死信キューへ隔離されます。
九、死信キューの名前は一般的にどのようになるか？
RocketMQ の死信トピックは通常次のような形式です:
%DLQ%consumer_group
例えばコンシューマーグループが:
inventory_consumer_group
なら、死信キューは:
%DLQ%inventory_consumer_group
になります。
つまり:
死信キューは Consumer Group の単位で生成されます。
なぜか？
同じメッセージでも、コンシューマーグループによって消費結果が異なる可能性があるからです。
例えば:
order_topic に 1 件の注文メッセージがある
在庫システムは失敗:
inventory_group は消費失敗
SMS システムは成功:
sms_group は消費成功
すると、このメッセージは inventory_group にだけ死信キューとして入り、sms_group には影響しません。
十、完全な例
注文サービスがメッセージを送信:
Topic: order_topic
Tag: create
Body: {"orderId":10001}
在庫コンシューマー:
@Component
@RocketMQMessageListener(
topic = "order_topic",
consumerGroup = "inventory_group",
selectorExpression = "create"
)
public class InventoryConsumer implements RocketMQListener<String> {
@Override
public void onMessage(String payload) {
// ずっと失敗することをシミュレート
throw new RuntimeException("在庫減算に失敗");
}
}
RocketMQ は繰り返し再試行します。
ずっと失敗すると、最終的にメッセージは次に入ります:
%DLQ%inventory_group
その後、次のことができます:
死信メッセージを確認する
失敗原因を分析する
データまたはコードを修正する
メッセージを再送信する
十一、本当に覚えておくべきポイント
プロデューサー送信の信頼性
同期送信 + SEND_OK の確認
これでメッセージが Broker に正常に到達したことを保証できます。
コンシューマー消費の信頼性
業務処理成功 -> 正常に戻す
業務処理失敗 -> 例外を投げて RocketMQ に再試行させる
重複消費の防止
コンシューマーは必ず冪等にする
メッセージが重複配信される可能性があるためです。
死信キュー
メッセージの消費が何度も失敗すると死信キューに入り、人の調査や再送信を待つ。