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

インスタンスの合計 TPS スロットリングしきい値

シングルノードの SendMessage TPS スロットリングしきい値

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)

エラーコードを処理するためのサンプルコード

次のサンプルコードは 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("Failed to create channel. Attempt " + attempt + " of " + MAX_RETRIES);
            // エラーを確認します。スロットリングが原因でチャンネルがまだ閉じている場合は、待機してからリトライできます。
            // このリトライロジックを削除することもできます。
            if (attempt < MAX_RETRIES) {
                try {
                    Thread.sleep(WAIT_TIME_MS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // 中断された状態を復元します。
                }
            } else {
                throw new RuntimeException("Exceeded maximum retries to create channel", e);
            }
        }
    }
    throw new RuntimeException("This line should never be reached"); // 理論上、このコード行には到達できません。
}

private boolean isChannelClosed(String errorMsg) {
    // エラーメッセージに "channel.close" が含まれているかどうかを確認します。このエラーは、チャンネルが閉じていることを示します。
    // エラーには、530 や 541 などのエラーメッセージが含まれる場合があります。
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        System.out.println("[ChannelClosed] Error details: " + errorMsg);
        return true;
    }
    return false;
}

単一インスタンスの合計 TPS 超過によるスロットリングのソリューション

• テストシナリオ、またはピークトラフィックが不確定またはトラフィック量が少ない短期的なシナリオでは、サーバーレスの ApsaraMQ for RabbitMQ インスタンスを使用することをお勧めします。このようなシナリオでサブスクリプションベースの ApsaraMQ for RabbitMQ インスタンスを使用する場合は、インスタンスの Elastic TPS 機能を有効にすることをお勧めします。詳細については、「インスタンスの Elastic TPS 機能を有効にする」をご参照ください。

• トラフィックが安定して大量にある長期的なシナリオでは、TPS 仕様をスペックアップすることをお勧めします。詳細については、「インスタンス構成のスペックアップ」をご参照ください。

単一ノードでの TPS 超過によるスロットリングのソリューション

• ApsaraMQ for RabbitMQ クラスターは分散アーキテクチャを使用します。クライアントがクラスター内の複数のサービスノードにバランスよく接続できるように、各キューに複数の接続 (少なくとも 10) を作成することをお勧めします。このメソッドは、負荷のホットスポットを効果的に防止し、メッセージの送信と消費の効率を向上させることができます。

• Spring ユーザーの場合は、CachingConnectionFactory の CONNECTION モードを使用することをお勧めします。詳細については、「Spring 統合」をご参照ください。