ApsaraMQ for RabbitMQ:インスタンスのスロットリングに関するベストプラクティス

インスタンス合計 TPS

トラフィックパターンに合わせて以下のアプローチを選択してください。

単一ノード SendMessage TPS

ApsaraMQ for RabbitMQ は、複数のバックエンドノードにトラフィックを分散します。トラフィックが 1 つのノードに集中すると、インスタンスレベルの制限に余裕があっても、ノードごとの制限が発動します。

ノード間に負荷を均等に分散するには、以下の対策を実施してください。

• キューごとに少なくとも 10 個の接続を開きます。各接続は異なるノードにルーティングされる可能性があり、送信ワークロードが分散されます。

• Spring ユーザーの場合は、 CachingConnectionFactory を CONNECTION モードに設定してください。これにより、1 つの接続でチャネルを多重化するのではなく、セッションごとに新しい接続が作成されます。詳細については、「Spring 統合」をご参照ください。

CloudMonitor (推奨)

• 粒度：分レベル (1 分間隔)

• 範囲：インスタンスレベルのピーク TPS

• コスト：無料

• 過去 14 日間の TPS トレンドを表示します。

• スロットリング発生前に通知するアラートルールをサポートします。

詳細については、「CloudMonitor を使用したインスタンスのピーク TPS のクエリとアラートルールの設定」をご参照ください。

インスタンス詳細ページ (推奨)

• 粒度：秒レベルのピーク TPS

• 範囲：インスタンスレベルおよび API 操作ごとのピーク TPS

• コスト：無料

• 秒レベルの精度で短時間のトラフィックスパイクを特定します。

• API 操作ごとのフィルタリングをサポートします。

最初の 10 分間の結果のみが表示されます。

詳細については、「インスタンス詳細ページでのインスタンスのピーク TPS のクエリ」をご参照ください。

Simple Log Service

• 粒度：秒レベルのピーク TPS

• 範囲：インスタンスレベルのピーク TPS

• コスト：Simple Log Service による課金 (詳細については、「従量課金の課金項目」をご参照ください)

• 高度な分析のための SLS クエリステートメントをサポートします。

SLS クエリ構文の知識が必要です。CloudMonitor やインスタンス詳細ページと比較すると、結果の解釈が難しい場合があります。

詳細については、「Simple Log Service を使用したインスタンスのピーク TPS のクエリ」をご参照ください。

インスタンス合計 TPS

単一ノード SendMessage TPS

ブローカーは、インスタンス内の各バックエンドノードで SendMessage TPS を制限します。

API 操作ごとの制限

これらの制限はインスタンスごとに適用されます。サーバーレス専用クラスターインスタンスには、API ごとのスロットリングはありません。

エラーコードとメッセージ

スロットリングが発動すると、ブローカーは影響を受けたチャネルを閉じ、以下を返します。

• エラーコード： reply-code=530

• エラーメッセージ： reply-text=denied for too many requests

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>
(reply-code=530, reply-text=denied for too many requests, ReqId:5FB4C999314635F952FCBFF6, ErrorHelp[dstQueue=XXX_test_queue,
srcExchange=Producer.ExchangeName,bindingKey=XXX_test_bk, http://mrw.so/6rNqO8], class-id=50, method-id=20)
    at com.rabbitmq.client.impl.ChannelN.asyncShutdown(ChannelN.java:516)
    at com.rabbitmq.client.impl.ChannelN.processAsync(ChannelN.java:346)
    at com.rabbitmq.client.impl.AMQChannel.handleCompleteInboundCommand(AMQChannel.java:182)
    at com.rabbitmq.client.impl.AMQChannel.handleFrame(AMQChannel.java:114)
    at com.rabbitmq.client.impl.AMQConnection.readFrame(AMQConnection.java:672)
    at com.rabbitmq.client.impl.AMQConnection.access$300(AMQConnection.java:48)
    at com.rabbitmq.client.impl.AMQConnection$MainLoop.run(AMQConnection.java:599)
    at java.lang.Thread.run(Thread.java:748)

エラーには、リクエスト ID (ReqId)、宛先キュー (dstQueue)、送信元 Exchange (srcExchange)、およびバインディングキー (bindingKey) が含まれます。

チャネル閉鎖からの復旧

チャネルのみが閉じられるため、AlreadyClosedException をキャッチして再作成してください。以下の Java サンプルは、リトライによる復旧を示しています。

private static final int MAX_RETRIES = 5; // 最大リトライ試行回数
private static final long WAIT_TIME_MS = 2000; // リトライ間の待機時間 (ミリ秒)

private void doAnythingWithReopenChannels(Connection connection, Channel channel) {
    try {
        // ......
        // 現在のチャネルで実行する任意の操作。
        // 例: メッセージの送信または消費。
        // ......

    } catch (AlreadyClosedException e) {
        String message = e.getMessage();
        if (isChannelClosed(message)) {
            // チャネルがブローカーによって閉じられました。再作成します。
            channel = createChannelWithRetry(connection);
            // 復旧後、他の操作を続行します。
            // ......
        } else {
            throw e;
        }
    }
}

private Channel createChannelWithRetry(Connection connection) {
    for (int attempt = 1; attempt <= MAX_RETRIES; attempt++) {
        try {
            return connection.createChannel();
        } catch (Exception e) {
            System.err.println("チャネルの作成に失敗しました。試行 " + attempt + " / " + MAX_RETRIES);
            // チャネル作成が失敗した場合 (おそらくまだスロットリング中)、待機してリトライします。
            if (attempt < MAX_RETRIES) {
                try {
                    Thread.sleep(WAIT_TIME_MS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // 割り込み状態を復元します。
                }
            } else {
                throw new RuntimeException("チャネル作成の最大リトライ回数を超えました", e);
            }
        }
    }
    throw new RuntimeException("この行には到達しないはずです");
}

private boolean isChannelClosed(String errorMsg) {
    // エラーメッセージに "channel.close" が含まれているかどうかを確認します。
    // これは、エラーコード 530 (スロットリング) と 541 (内部エラー) の両方をカバーします。
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        System.out.println("[ChannelClosed] エラー詳細: " + errorMsg);
        return true;
    }
    return false;
}

重要なポイント：

• リトライでは固定の 2,000 ミリ秒の待機時間を使用しています。トラフィックパターンに応じて調整してください。

• isChannelClosed メソッドは、エラーメッセージ内の channel.close をチェックします。これは、reply-code=530 (スロットリング) およびその他のチャネル閉鎖シナリオの両方をカバーします。

• 復旧後、新しいチャネルオブジェクトで操作を再開してください。