RabbitMQの高度な機能

1.1. プロデューサーの再試行メカニズム

まず、プロデューサーがメッセージを送信する際にネットワーク障害が発生し、MQとの接続が切断される場合があります。

この問題を解決するために、SpringAMQPはメッセージ送信時の再試行メカニズムを提供しています。つまり、RabbitTemplateがMQとの接続がタイムアウトした後、複数回再試行します。

publisherモジュールのapplication.yamlファイルを修正し、以下の内容を追加します：

spring:
  rabbitmq:
    connectionTimeout: 1s # MQの接続タイムアウト時間を設定
    template:
      retry:
        enabled: true # タイムアウト再試行メカニズムを有効化
        initialInterval: 1000ms # 失敗後の初期待機時間
        multiplier: 1.5 # 失敗後の次の待機時間倍数、次の待機時間 = initialInterval * multiplier
        maxAttempts: 5 # 最大再試行回数

コマンドを使用してRabbitMQサービスを停止します：

docker stop mq

次にメッセージを送信してテストすると、1秒ごとに再試行され、合計5回再試行されることがわかります。メッセージ送信のタイムアウト再試行メカニズムの設定が成功しました！

注意：ネットワークが不安定な場合、再試行メカニズムを利用することでメッセージ送信の成功率を効果的に向上させることができます。しかし、SpringAMQPが提供する再試行メカニズムはブロッキング型の再試行であり、複数回の再試行待機中に現在のスレッドがブロックされることを意味します。

ビジネスパフォーマンスに要求がある場合は、再試行メカニズムを無効にすることを推奨します。必ず使用する場合は、待機時間と再試行回数を適切に設定してください。もちろん、非同期スレッドを使用してメッセージ送信のコードを実行することも考慮できます。

1.3. プロデューサーの確認メカニズム

通常、プロデューサーとMQの間のネットワーク接続がスムーズであれば、メッセージ送信のロスは基本的に発生しません。したがって、ほとんどの場合、この問題を考慮する必要はありません。

しかし、稀なケースでは、メッセージがMQに送信された後にロスが発生する現象もあります。例えば：

• MQ内部のメッセージ処理プロセスに異常が発生
• プロデューサーがメッセージをMQに送信した後、Exchangeが見つからない
• プロデューサーがメッセージをMQのExchangeに送信した後、適切なQueueが見つからず、ルーティングできない

上記の状況に対処するために、RabbitMQはプロデューサーメッセージ確認メカニズムを提供しています。これにはPublisher ConfirmとPublisher Returnの2種類があります。確認メカニズムを有効にした場合、プロデューサーがMQにメッセージを送信した後、MQはメッセージ処理の状況に基づいて異なるレシートを返します。

具体的には以下の通りです：

• メッセージがMQに配信されたが、ルーティングに失敗した場合、Publisher Returnを通じて異常情報を返し、同時にackの確認情報を返し、配信成功を示します
• 一時メッセージがMQに配信され、キューに入隊に成功し、ACKを返し、配信成功を示します
• 永続メッセージがMQに配信され、キューに入隊して永続化が完了し、ACKを返し、配信成功を示します
• その他の状況ではNACKを返し、配信失敗を示します

ここでackとnackはPublisher Confirmメカニズムに属し、ackは配信成功を、nackは配信失敗を示します。一方、returnはPublisher Returnメカニズムに属します。

デフォルトでは両方のメカニズムが無効状態であり、設定ファイルを通じて有効にする必要があります。

設定

RabbitMqConfig.java

@Configuration
@Slf4j
@RequiredArgsConstructor
public class RabbitMqConfig {

    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void initialize() {
        rabbitTemplate.setReturnsCallback(returnedMessage -> {
            log.error("リターンコールバックがトリガーされました");
            log.debug("エクスチェンジ: {}", returnedMessage.getExchange());
            log.debug("ルーティングキー: {}", returnedMessage.getRoutingKey());
            log.debug("メッセージ: {}", returnedMessage.getMessage());
            log.debug("リプライコード: {}", returnedMessage.getReplyCode());
            log.debug("リプライテキスト: {}", returnedMessage.getReplyText());
        });
    }
}

メッセージ送信：

@Test
public void testConfirmCallback() throws InterruptedException {
    // 0. CorrelationDataを作成
    CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());
    cd.getFuture().addCallback(new ListenableFutureCallback() {
        @Override
        public void onFailure(Throwable ex) {
            log.info("Spring AMQPの確認処理で例外が発生しました", ex);
        }

        @Override
        public void onSuccess(CorrelationData.Confirm result) {
            // 成功かどうかを判断
            if (result.isAck()) {
                log.debug("ConfirmCallbackのACKを受信しました。メッセージ送信成功");
            } else {
                log.debug("ConfirmCallbackのNACKを受信しました。メッセージ送信失敗: {}", result.getReason());
            }
        }
    });

    // 1. キュー名
    String exchangeName = "hmall.direct";
    // 2. メッセージ
    String message = "今日は天気が非常に良いです";
    // 3. メッセージを送信
    rabbitTemplate.convertAndSend(exchangeName, "blue", message, cd);

    Thread.sleep(2);
}

1.3.2. ReturnCallbackの定義

各RabbitTemplateには1つのReturnCallbackしか設定できないため、設定クラスで統一して設定できます。publisherモジュールで設定クラスを定義します：

内容は以下の通り：

package com.itheima.publisher.config;

import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Slf4j
@AllArgsConstructor
@Configuration
public class RabbitMqConfig {
    private final RabbitTemplate rabbitTemplate;

    @PostConstruct
    public void initialize(){
        rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
            @Override
            public void returnedMessage(ReturnedMessage returned) {
                log.error("リターンコールバックがトリガーされました");
                log.debug("エクスチェンジ: {}", returned.getExchange());
                log.debug("ルーティングキー: {}", returned.getRoutingKey());
                log.debug("メッセージ: {}", returned.getMessage());
                log.debug("リプライコード: {}", returned.getReplyCode());
                log.debug("リプライテキスト: {}", returned.getReplyText());
            }
        });
    }
}

1.3.3. ConfirmCallbackの定義

各メッセージ送信時の処理ロジックが異なる可能性があるため、ConfirmCallbackは各メッセージ送信時に定義する必要があります。具体的には、RabbitTemplateのconvertAndSendメソッドを呼び出す際に、追加のパラメータを渡します：

ここでのCorrelationDataには2つの核心的なものが含まれています：

• id：メッセージの一意の識別子、MQは異なるメッセージのレシートをこの識別子で判断し、混同を避けます
• SettableListenableFuture：レシート結果のFutureオブジェクト

将来、MQのレシートはこのFutureを通じて返され、事前にCorrelationDataのFutureにコールバック関数を追加してメッセージレシートを処理できます：

新しいテストを作成し、システムのデフォルトエクスチェンジにメッセージを送信し、ConfirmCallbackを追加します：

@Test
void testPublisherConfirm() {
    // 1. CorrelationDataを作成
    CorrelationData cd = new CorrelationData();
    // 2. FutureにConfirmCallbackを追加
    cd.getFuture().addCallback(new ListenableFutureCallback() {
        @Override
        public void onFailure(Throwable ex) {
            // 2.1. Futureが例外を発生した場合の処理ロジック、基本的にトリガーされない
            log.error("メッセージ送信に失敗しました", ex);
        }
        @Override
        public void onSuccess(CorrelationData.Confirm result) {
            // 2.2. Futureがレシートを受信した場合の処理ロジック、パラメータのresultがレシート内容
            if(result.isAck()){ // result.isAck()はboolean型、trueはackレシート、falseはnackレシートを示す
                log.debug("メッセージ送信成功、ACKを受信しました");
            } else { // result.getReason()はString型、nack時の例外記述を返す
                log.error("メッセージ送信失敗、NACKを受信しました。理由: {}", result.getReason());
            }
        }
    });
    // 3. メッセージを送信
    rabbitTemplate.convertAndSend("hmall.direct", "q", "hello", cd);
}

実行結果は以下の通り：

渡されたRoutingKeyが間違っているため、ルーティング失敗後にreturn callbackがトリガーされ、同時にackも受信しました。

正しいRoutingKeyに修正すると、return callbackはトリガーされず、ackのみ受信します。

そして、エクスチェンジが間違っている場合は、nackのみ受信します。

2. MQの信頼性

メッセージがMQに到達した後、MQが適時に保存できない場合、メッセージロスが発生する可能性があります。したがって、MQの信頼性も非常に重要です。

2.1. データの永続化

パフォーマンスを向上させるために、デフォルトではMQのデータはメモリに保存される一時データです。再起動後に消失します。データの信頼性を保証するために、データの永続化を設定する必要があります。これには以下が含まれます：

• エクスチェンジの永続化
• キューの永続化
• メッセージの永続化

コンソールインターフェースを例に説明します。

2.1.1. エクスチェンジの永続化

コンソールのExchangesページで、エクスチェンジを追加する際にエクスチェンジのDurabilityパラメータを設定できます：

設定がDurableの場合は永続化モード、Transientの場合は一時モードです。

2.1.2. キューの永続化

コンソールのQueuesページで、同様にキューのDurabilityパラメータを設定できます：

2.1.3. メッセージの永続化

コンソールでメッセージを送信する際に、多くのパラメータを追加できます。メッセージの永続化はpropertiesを設定する必要があります：

.setDeliveryMode(MessageDeliveryMode.PERSISTENT)をカスタム永続化として設定します。

2.2. LazyQueue

デフォルトでは、RabbitMQは受信した情報をメモリに保存して**メッセージの送受信遅延を低減**します。しかし、**特定の状況では、これがメッセージの蓄積を引き起こす**可能性があります。例えば：

• コンシューマーがダウンまたはネットワーク障害が発生
• メッセージ送信量が急増し、コンシューマーの処理速度を超える
• コンシューマーがビジネス処理でブロックされる

メッセージ蓄積問題が発生すると、RabbitMQのメモリ使用量がますます高くなり、メモリ警告上限に達するまでになります。この時点でRabbitMQはメモリメッセージをディスクにフラッシュします。この動作はPageOutと呼ばれます。PageOutは一定時間を要し、キューのプロセスをブロックします。したがって、このプロセス中、RabbitMQは新しいメッセージを処理せず、プロデューサーのすべてのリクエストがブロックされます。

この問題を解決するために、RabbitMQの3.6.0バージョンから、Lazy Queuesのモードが追加されました。Lazy Queuesの特徴は以下の通りです：

• メッセージを受信した後、メモリではなくディスクに直接保存
• コンシューマーがメッセージを消費する際に、ディスクから読み込みメモリにロード（遅延ロード）
• 数百万件のメッセージの保存をサポート

3.12バージョン以降、LazyQueueはすべてのキューのデフォルト形式になりました。したがって、MQを3.12バージョンにアップグレードするか、すべてのキューをLazyQueueモードに設定することを公式が推奨しています。

2.2.1. コンソールでLazyモードを設定

キューを追加する際に、x-queue-mod=lazyパラメータを追加してLazyモードに設定できます：

2.2.2. コードでLazyモードを設定

SpringAMQPを使用してキューを宣言する際に、x-queue-mod=lazyパラメータを追加してLazyモードに設定できます：

@Bean
public Queue lazyQueue(){
    return QueueBuilder
            .durable("lazy.queue")
            .lazy() // Lazyモードを有効化
            .build();
}

ここではQueueBuilderのlazy()関数を使用してLazyモードを設定します。下層のソースコードは以下の通りです：

もちろん、注釈に基づいてキューを宣言し、Lazyモードを設定することもできます：

@RabbitListener(queuesToDeclare = @Queue(
        name = "lazy.queue",
        durable = "true",
        arguments = @Argument(name = "x-queue-mode", value = "lazy")
))
public void listenLazyQueue(String msg){
    log.info("lazy.queueのメッセージを受信しました：{}", msg);
}

2.2.3. 既存のキューをlazyモードに更新

既存のキューもlazyモードに設定できますが、policyを設定する必要があります。

コマンドラインに基づいてpolicyを設定できます：

rabbitmqctl set_policy Lazy "^lazy-queue$" '{"queue-mode":"lazy"}' --apply-to queues

コマンドの解説：

• rabbitmqctl：RabbitMQのコマンドラインツール
• set_policy：ポリシーを追加
• Lazy：ポリシー名、カスタマイズ可能
• "^lazy-queue$"：正規表現を使用してキュー名をマッチング
• '{"queue-mode":"lazy"}'：キューのモードをlazyに設定
• --apply-to queues：ポリシーの対象、すべてのキュー

もちろん、コンソールでもpolicyを設定できます。コンソールのAdminページに進み、Policiesをクリックして設定を追加します：

3. コンシューマーの信頼性

2.1. コンシューマーの確認メカニズム

2.2. 失敗再試行メカニズム

1）consumerサービスで失敗メッセージを処理するエクスチェンジとキューを定義します

@Bean
public DirectExchange errorMessageExchange(){
    return new DirectExchange("error.direct");
}
@Bean
public Queue errorQueue(){
    return new Queue("error.queue", true);
}
@Bean
public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
    return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
}

2）RepublishMessageRecovererを定義し、キューとエクスチェンジを関連付けます

@Bean
public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
    return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
}

package com.itheima.consumer.config;

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.retry.MessageRecoverer;
import org.springframework.amqp.rabbit.retry.RepublishMessageRecoverer;
import org.springframework.context.annotation.Bean;

@Configuration
@ConditionalOnProperty(name = "spring.rabbitmq.listener.simple.retry.enabled", havingValue = "true")
public class ErrorMessageConfig {
    @Bean
    public DirectExchange errorMessageExchange(){
        return new DirectExchange("error.direct");
    }
    @Bean
    public Queue errorQueue(){
        return new Queue("error.queue", true);
    }
    @Bean
    public Binding errorBinding(Queue errorQueue, DirectExchange errorMessageExchange){
        return BindingBuilder.bind(errorQueue).to(errorMessageExchange).with("error");
    }

    @Bean
    public MessageRecoverer republishMessageRecoverer(RabbitTemplate rabbitTemplate){
        return new RepublishMessageRecoverer(rabbitTemplate, "error.direct", "error");
    }
}