云消息队列 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集成。