ApsaraMQ for RabbitMQ：实例限流最佳实践

執行個體總TPS限流閾值

單節點SendMessage TPS限流閾值

單介面的限流閾值

錯誤碼資訊

• 錯誤碼：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 {
        // ......
        // 在當前通道channel下執行的任何操作
        // 例如訊息發送、消費等
        // ......

    } 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被限流的解決辦法

• 如果在測試或者流量峰值不確定、流量較少的短期情境，建議您為預付費系列執行個體開啟彈性TPS能力或者使用Serverless系列執行個體。更多資訊，請參見為執行個體開啟彈性TPS功能。

• 如果在長期穩定、流量較高的業務運行情境，建議您升級TPS流量峰值規格。更多資訊，請參見升級執行個體配置。

單節點的TPS被限流的解決辦法

• 雲訊息佇列 RabbitMQ 版採用分布式叢集架構。建議為每個隊列建立多個串連（至少10個），以便用戶端能夠更均衡地串連到叢集中的多個服務節點。這種方法可以有效避免出現負載熱點問題，從而提高訊息的發送和消費效率。

• Spring使用者推薦使用CachingConnectionFactory的CONNECTION模式，詳情請參見Spring整合。