ApsaraMQ for RabbitMQ:オープンソース RabbitMQ クライアントの使用上の注意

自動接続復旧の有効化

com.rabbitmq.client.ConnectionFactory で自動接続復旧を有効にします。有効にしない場合、ブローカーのアップグレード時にメッセージの読み書きが中断され、クライアントが自動的に再接続できなくなります。

// 自動接続復旧を有効にします
factory.setAutomaticRecoveryEnabled(true);
// 復旧試行間隔を設定します (ミリ秒)
factory.setNetworkRecoveryInterval(5000);

永続的な接続の使用

パブリッシュ操作や消費操作のたびに接続を開閉しないでください。頻繁な再接続は、ネットワークリソースとブローカーリソースを浪費し、SYN フラッド保護をトリガーする可能性があります。

詳細については、「接続とチャネル」をご参照ください。

アプリケーションコード内でのメタデータ操作の回避

queueDeclare や exchangeDeclare などのメタデータ操作は、スロットリングの対象となります。スロットリングがトリガーされると、ブローカーが接続を閉じる可能性があります。

キューとエクスチェンジは、メッセージ送信時の実行時に宣言するのではなく、ApsaraMQ for RabbitMQ コンソールを通じて作成してください。スロットリング制限の詳細については、「制限」をご参照ください。

パブリッシャー確認の設定

パブリッシャー確認を使用すると、ブローカーが各メッセージをディスクに書き込んだ後に確認応答を返すため、クライアントは配信の失敗を検出して再試行できます。この確認ステップにより、メッセージが永続化されたことを保証できますが、ブローカーからの応答を待つため、レイテンシが増加する可能性があります。

チャネルでパブリッシャー確認を有効にするには、次の手順に従います：

1. チャネル作成時に confirmSelect() を呼び出します。

   // 接続とチャネルを作成します
   Connection connection = createConnection(hostName, userName, passWord, virtualHost);
   Channel channel = connection.createChannel();
   // パブリッシャー確認を有効にします
   channel.confirmSelect();

2. basicPublish の呼び出し後、ブローカーの応答を待機し、結果を処理します。

   // メッセージをパブリッシュします (既存のチャネルを再利用)
   channel.basicPublish(exchangeName, bindingKey, true, props,
                   ("example body").getBytes(StandardCharsets.UTF_8));
   
   // 最大 3 秒間確認を待機します
   if (channel.waitForConfirms(3000)) {
       // メッセージがディスクへ正常に永続化されました
   } else {
       // 確認に失敗しました -- メッセージを再送信します
   }

mandatory フラグによるルーティングできないメッセージの処理

mandatory パラメータを true に設定すると、メッセージがどのキューにもルーティングできない場合に、ブローカーが ReturnListener コールバックを呼び出します。

// ルーティングできないメッセージ用のコールバックを登録します
channel.addReturnListener(returnMessage ->
    System.out.println("return msgId=" + returnMessage.getProperties().getMessageId()));
// mandatory=true でパブリッシュします (第 3 パラメータ)
channel.basicPublish(exchangeName, routingKey, true, props,
    content.getBytes(StandardCharsets.UTF_8));

すべてのメッセージへのメッセージ ID の割り当て

各メッセージにカスタム ID を割り当てることで、トレースをクエリし、配信の問題をトラブルシューティングできます。詳細については、「メッセージ ID を指定するにはどうすればよいですか?」をご参照ください。

パブリッシュエラーの処理

basicPublish が返すエラータイプに基づいて、エラー処理戦略を決定してください。

• ビジネス例外 (ExchangeNotExist など)： 直ちに例外をスローします。

• スロットリングまたはチャネルクローズ： 既存の接続を閉じ、新しいチャネルを作成して初期化し、送信を再試行します。

例：

private void doSend(String content) throws Exception {
    try {
        // 一意のメッセージ ID を割り当てます
        String msgId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();

        channel.basicPublish(exchangeName, routingKey, true, props, content.getBytes(StandardCharsets.UTF_8));
        // 最大 3 秒間確認を待機します
        if (channel.waitForConfirms(3000)) {
            // メッセージがディスクへ正常に永続化されました
        } else {
            // 確認に失敗しました -- メッセージを再送信します
        }

    } catch (Exception e) {
        String message = e.getMessage();
        System.out.println(message);

        if (channelClosedByServer(message)) {
            // 復旧: 古いチャネルを閉じ、新しいチャネルを作成して再初期化し、再試行します
            factory.closeCon(channel);
            channel = factory.createChannel();
            this.initChannel();
            doSend(content);
        } else {
            // ブローカー以外のエラー -- 例外を伝播します
            throw e;
        }
    }
}

private boolean channelClosedByServer(String errorMsg) {
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        return true;
    } else {
        return false;
    }
}

完全なサンプルコードについては、「メッセージの送信」をご参照ください。

Spring Boot クライアントの接続モードの設定

Spring Boot クライアントを使用する場合、接続キャッシュモードをデフォルトの channel モードではなく、connection に明示的に設定する必要があります。接続キャッシングをサポートする他のクライアントライブラリについても同じ原則が適用されます。常に接続モードを明示的に設定してください。

たとえば、application.properties または application.yml で次のプロパティを設定します：

spring.rabbitmq.cache.connection.mode=connection

たとえば、application.properties または application.yml に次のプロパティを設定します：

spring.rabbitmq.cache.connection.mode=connection

プロデューサーごとの複数接続の確立

メッセージを送信する際、各プロデューサークライアントは複数の接続を確立して、複数のバックエンドサービスノードに接続する必要があります。これにより、単一の接続における突発的なトラフィックスパイクによるリソースのボトルネックを回避できます。

推奨事項：

• 各プロデューサークライアントは、バックエンドに 50 接続を確立する必要があります。

• これらの接続は、複数のバックエンドサービスノードに分散させる必要があります。

消費の偏りの防止

メッセージがコンシューマー間で均等に分散されるようにしてください。消費の偏りが発生すると、一部のコンシューマーが過負荷になり、他のコンシューマーがアイドル状態になります。設定の詳細については、「接続とチャネル」の「使用上の注意」セクションをご参照ください。

ブローカー生成のコンシューマータグの使用

手動でコンシューマータグを指定するのではなく、ブローカーにグローバルに一意なコンシューマータグを生成させてください。カスタムコンシューマータグを指定する場合は、すべてのコンシューマー間で一意であることを確認してください。

QoS (プリフェッチ) の設定

basicQos を呼び出して、ブローカーがコンシューマーにプッシュする未確認のメッセージ数を制限します。この制限に達すると、コンシューマーが未処理のメッセージを確認するまで、ブローカーは配信を一時停止します。

スコープオプション：

プラットフォームの制約：

• デフォルトの QoS 値は、コンシューマーあたり 100 で、channel.basicQos(100, false) と同等です。

• カスタム QoS 値は 100 を超えることはできません。100 を超える値は無視され、代わりにデフォルト値が使用されます。

• basicQos は、自動 ACK モードでは効果がありません。

QoS 値の選択：

ACK タイムアウトとリトライ動作

コンシューマーが設定されたタイムアウト内にメッセージを確認しない場合、ブローカーはメッセージを再配信します。

各メッセージは、設定された最大回数まで再試行できます。最大再試行回数に達した後、メッセージは破棄されるか、デッドレターエクスチェンジにルーティングされます。

タイムアウトと再試行制限は、インスタンスタイプとエディションによって異なります。

basicGet ではなく basicConsume の使用

basicGet は個別のメッセージをポーリングし、basicConsume と比較してスループットとトランザクション/秒 (TPS) の制限が低くなります。メッセージ量が多い本番環境では、プッシュベースの配信を行うには、長時間稼働するコンシューマーで basicConsume を使用してください。