ApsaraMQ for RabbitMQ:Best practices for instance throttling

Instance total TPS throttling threshold

Single-node SendMessage TPS throttling threshold

API operation throttling threshold

Error code and error message

• Error code: reply-code=530

• Error message: 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)

Sample code for handling error codes

The following sample code is in Java:

private static final int MAX_RETRIES = 5; // The maximum number of retries.
private static final long WAIT_TIME_MS = 2000; // The wait time for each retry in milliseconds.

private void doAnythingWithReopenChannels(Connection connection, Channel channel) {
    try {
        // ......
        // Any operation to be performed in the current channel.
        // For example, sending or consuming messages.
        // ......

    } catch (AlreadyClosedException e) {
        String message = e.getMessage();
        if (isChannelClosed(message)) {
            // If the channel is closed, close and re-create the channel.
            channel = createChannelWithRetry(connection); 
            // You can continue to perform other operations after reconnection.
            // ......
        } 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);
            // Check for errors. If the channel is still closed due to throttling, you can wait and then retry.
            // You can also remove this retry logic.
            if (attempt < MAX_RETRIES) {
                try {
                    Thread.sleep(WAIT_TIME_MS);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt(); // Restore the interrupted state.
                }
            } else {
                throw new RuntimeException("Exceeded maximum retries to create channel", e);
            }
        }
    }
    throw new RuntimeException("This line should never be reached"); // In theory, this line of code is unreachable.
}

private boolean isChannelClosed(String errorMsg) {
    // Check whether the error message contains "channel.close". This error indicates that the channel is closed.
    // The error may contain error messages such as 530 and 541.
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        System.out.println("[ChannelClosed] Error details: " + errorMsg);
        return true;
    }
    return false;
}

Solutions to throttling caused by the exceeding of the total TPS of a single instance

• For test scenarios or short-term scenarios with uncertain peak traffic or small amounts of traffic, we recommend that you use a serverless ApsaraMQ for RabbitMQ instance. If you use a subscription-based ApsaraMQ for RabbitMQ instance in such scenarios, we recommend that you enable the elastic TPS feature for the instance. For more information, see Enable the elastic TPS feature for an instance.

• For long-term scenarios with stable and large amounts of traffic, we recommend that you upgrade the TPS specification. For more information, see Upgrade instance configurations.

Solutions to throttling caused by the exceeding of TPS on a single node

• An ApsaraMQ for RabbitMQ cluster uses a distributed architecture. We recommend that you create multiple connections (at least 10) for each queue so that clients can connect to multiple service nodes in the cluster in a balanced manner. This method can effectively prevent load hotspots and improve message sending and consumption efficiency.

• If you are a Spring user, we recommend that you use the CONNECTION mode of CachingConnectionFactory. For more information, see Spring integration.