RabbitMQ

Ack / Nack / Reject の実装とベストプラクティス

basicAck

用途(ようと) ：「このメッセージは正常(せいじょう) に消費(しょうひ) され処理(しょり) が完了(かんりょう) した」ことを示(しめ) す。

構文(こうぶん) ：channel.basicAck(deliveryTag, multiple)

パラメータ	説明(せつめい)
`deliveryTag`	メッセージの一意(いちい) 識別子(しきべつし) （RabbitMQ が割(わ) り当(あ) て）
`multiple`	`true` の場合(ばあい) 、deliveryTag 以下(いか) のすべてのメッセージを一括(いっかつ) ack

channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

ベストプラクティス

通常(つうじょう) 、ビジネスロジックが成功(せいこう) した後(あと) に Ack することを推奨(すいしょう)
Spring AMQP を使用(しよう) する場合(ばあい) 、ackMode = MANUAL と channel.basicAck() を組(く) み合(あ) わせる
try 外(そと) で ack しないようにし、エラー時(じ) のメッセージ損失(そんしつ) を防(ふせ) ぐ

basicNack

用途(ようと) ：broker に「処理(しょり) 失敗(しっぱい) 」を明示的(めいじてき) に伝(つた) え、再配信(さいはいしん) するかどうかを選択(せんたく) できる。

構文(こうぶん) ：channel.basicNack(deliveryTag, multiple, requeue)

パラメータ	説明(せつめい)
`requeue = true`	キューに戻(もど) す（通常(つうじょう) 、重複(じゅうふく) 消費(しょうひ) が発生(はっせい) ）
`requeue = false`	破棄(はき) または DLX（デッドレターキュー）に転送(てんそう)

channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);

ベストプラクティス

ビジネスが再試行(さいしこう) 可能(かのう) な場合(ばあい) 、requeue = true を設定(せってい)
データ異常(いじょう) （フォーマットエラー、回復(かいふく) 不可能(ふかのう) ）を検出(けんしゅつ) した場合(ばあい) 、requeue = false → DLX で処理(しょり)
無限(むげん) requeue（メッセージストーム）を避(さ) け、retry 機構(きこう) や DLX で回数(かいすう) を制限(せいげん)

basicReject

用途(ようと) ：単一(たんいつ) のメッセージを拒否(きょひ) （複数(ふくすう) を一度(いちど) に拒否(きょひ) できない）。

構文(こうぶん) ：channel.basicReject(deliveryTag, requeue)

channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);

ベストプラクティス

機能(きのう) は basicNack に類似(るいじ) しているが、単一(たんいつ) メッセージのみ対象(たいしょう)
通常(つうじょう) 、basicNack の方(ほう) が機能(きのう) が豊富(ほうふ) なため、basicNack の使用(しよう) を推奨(すいしょう)

推奨の実装テンプレート

@RabbitListener(queues = "task.queue", ackMode = "MANUAL")
public void onMessage(Message message, Channel channel) throws IOException {
    long tag = message.getMessageProperties().getDeliveryTag();
    try {
        String body = new String(message.getBody());
        // ビジネスロジックを処理
        processTask(body);

        // 成功 → Ack
        channel.basicAck(tag, false);
    } catch (BusinessException e) {
        // 再試行可能なエラー
        channel.basicNack(tag, false, true);
    } catch (Exception e) {
        // 回復不可能なエラー → デッドレターキューへ
        channel.basicReject(tag, false);
    }
}

Retry 機構(きこう) 付(つ) きバージョン：

@RabbitListener(queues = "task.queue", ackMode = "MANUAL")
public void onMessage(Message message, Channel channel, CustomMessage payload) throws IOException {
    long tag = message.getMessageProperties().getDeliveryTag();
    Map<String, Object> headers = message.getMessageProperties().getHeaders();
    int retryCount = (int) headers.getOrDefault("x-retry-count", 0);

    try {
        // ビジネス層 payload でロジックを実行
        processBusinessLogic(payload);

        // 成功 → Ack
        channel.basicAck(tag, false);
    } catch (TemporaryException e) {
        // 一時的なエラー → Retry 機構
        if (retryCount < 3) {
            Message newMessage = MessageBuilder
                .withBody(message.getBody())
                .copyHeaders(headers)
                .setHeader("x-retry-count", retryCount + 1)
                .build();

            rabbitTemplate.convertAndSend(
                message.getMessageProperties().getReceivedExchange(),
                message.getMessageProperties().getReceivedRoutingKey(),
                newMessage
            );
        } else {
            log.error("Message {} failed 3 times: {}", payload.getTaskId(), e.getMessage());
        }
        channel.basicAck(tag, false);
    } catch (Exception e) {
        // 非一時的なエラー → 破棄
        log.error("Fatal error on message: {}", payload, e);
        channel.basicReject(tag, false);
    }
}

よく使う Class / Method / Property 一覧表

クラス / コンポーネント	属性(ぞくせい) / メソッド	説明(せつめい)
`Channel`	`basicAck(long tag, boolean multiple)`	手動(しゅどう) でメッセージ処理(しょり) 成功(せいこう) を確認(かくにん)
`Channel`	`basicNack(long tag, boolean multiple, boolean requeue)`	手動(しゅどう) 拒否(きょひ) （キューに戻(もど) すか選択(せんたく) 可能(かのう) ）
`Channel`	`basicReject(long tag, boolean requeue)`	単一(たんいつ) メッセージを拒否(きょひ)
`Channel`	`basicConsume(...)`	キューからメッセージの消費(しょうひ) を開始(かいし)
`Message`	`getBody()`	メッセージ内容(ないよう) を取得(しゅとく) （byte[]）
`Message`	`getMessageProperties()`	メッセージ属性(ぞくせい) オブジェクトを取得(しゅとく)
`MessageProperties`	`getDeliveryTag()`	delivery tag（メッセージ一意(いちい) 識別子(しきべつし) ）を取得(しゅとく)
`MessageProperties`	`getHeaders()`	カスタム headers（metadata）
`MessageProperties`	`getCorrelationId()`	Request-Response パターン用(よう)
`MessageProperties`	`getConsumerQueue()`	メッセージが属(ぞく) するキュー
`@RabbitListener`	`ackMode`	Ack モードを設定(せってい) ：`AUTO`、`MANUAL`、`NONE`
`AcknowledgeMode`	`AUTO / MANUAL / NONE`	コンシューマー確認(かくにん) モード
`AmqpRejectAndDontRequeueException`	例外(れいがい) クラス	拒否(きょひ) して再(さい) キューしない標準(ひょうじゅん) 例外(れいがい)
`RabbitTemplate`	`convertAndSend(exchange, routingKey, message)`	指定(してい) の Exchange にメッセージを送信(そうしん)
`SimpleMessageListenerContainer`	`setAcknowledgeMode(AcknowledgeMode.MANUAL)`	手動(しゅどう) ack モードを設定(せってい)

Ack モード比較

モード	動作(どうさ)	適用(てきよう) シナリオ
`AUTO`	処理(しょり) 成功(せいこう) 後(ご) に自動(じどう) ack、例外(れいがい) を投(な) げると自動(じどう) nack して requeue	ビジネスロジックが単純(たんじゅん)
`MANUAL`	`basicAck` / `basicNack` を自分(じぶん) で呼(よ) び出(だ) す必要(ひつよう) あり	複雑(ふくざつ) なビジネス、正確(せいかく) なリトライ制御(せいぎょ) が必要(ひつよう)
`NONE`	ack を返(かえ) さない、メッセージが重複(じゅうふく) する可能性(かのうせい) あり	非常(ひじょう) に稀(まれ) 、テスト用(よう) のみ

実務上の推奨事項（ベストプラクティスチェックリスト）

カテゴリ	推奨(すいしょう)
リトライ制御(せいぎょ)	Spring RetryTemplate またはカスタム retry 回数(かいすう) を使用(しよう) し、無限(むげん) requeue を避(さ) ける
デッドレターキュー (DLX)	消費(しょうひ) できないメッセージを処理(しょり) するデッドレターエクスチェンジを設定(せってい)
冪等性(べきとうせい)	消費(しょうひ) ロジックで重複(じゅうふく) メッセージが副作用(ふくさよう) を起(お) こさないことを確認(かくにん)
ロギング & トレーシング	`deliveryTag`、`messageId`、`correlationId` を記録(きろく) して追跡(ついせき) を容易(ようい) に
一括(いっかつ) Ack	`basicAck(tag, true)` でパフォーマンス向上(こうじょう) （データ整合性(せいごうせい) に注意(ちゅうい) ）
モニタリング	RabbitMQ Management Plugin で unacked 状態(じょうたい) を監視(かんし)

Message Queues 概要