ApsaraMQ for RabbitMQ:Best practices for instance throttling

Instance total TPS

Choose the approach that matches your traffic pattern:

Single-node SendMessage TPS

ApsaraMQ for RabbitMQ distributes traffic across multiple backend nodes. If traffic concentrates on one node, the per-node limit triggers even when the instance-level limit has headroom.

To distribute load evenly across nodes:

• Open at least 10 connections per queue. Each connection may route to a different node, distributing the send workload.

• For Spring users, set CachingConnectionFactory to CONNECTION mode. This creates a new connection per session instead of multiplexing channels on one. Spring integration.

CloudMonitor (recommended)

• Granularity: minute-level (1-minute period).

• Scope: instance-level peak TPS.

• Cost: free.

• Shows TPS trends over the past 14 days.

• Supports alert rules to notify before throttling occurs.

Query the peak TPS of an instance and configure an alert rule by using CloudMonitor.

Instance Details page (recommended)

• Granularity: second-level peak TPS.

• Scope: instance-level and per-API-operation peak TPS.

• Cost: free.

• Pinpoints short traffic spikes with second-level precision.

• Supports per-API-operation filtering.

Only the first 10 minutes of results are displayed.

Query the peak TPS of an instance on the Instance Details page.

Simple Log Service

• Granularity: second-level peak TPS.

• Scope: instance-level peak TPS.

• Cost: billed by Simple Log Service (Billable items of pay-by-feature).

• Supports SLS query statements for advanced analysis.

Requires familiarity with SLS query syntax. Results may be harder to interpret than CloudMonitor or Instance Details.

Query the peak TPS of an instance by using Simple Log Service.

Instance total TPS

Single-node SendMessage TPS

The broker limits SendMessage TPS on each backend node in the instance.

Per-API operation limits

These limits apply per instance. Serverless exclusive cluster instances have no per-API throttling.

Error code and message

When throttling is triggered, the broker closes the affected channel and returns:

• 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)

The error includes the request ID (ReqId), destination queue (dstQueue), source exchange (srcExchange), and binding key (bindingKey).

Recover from channel closure

Because only the channel is closed, catch AlreadyClosedException and recreate it. The following Java example demonstrates retry-based recovery:

private static final int MAX_RETRIES = 5; // Maximum retry attempts
private static final long WAIT_TIME_MS = 2000; // Wait time between retries 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)) {
            // Channel was closed by the broker. Recreate it.
            channel = createChannelWithRetry(connection);
            // Continue with other operations after recovery.
            // ......
        } 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 channel creation fails (possibly still throttled), wait and retry.
            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");
}

private boolean isChannelClosed(String errorMsg) {
    // Check whether the error message contains "channel.close".
    // This covers both error code 530 (throttling) and 541 (internal error).
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        System.out.println("[ChannelClosed] Error details: " + errorMsg);
        return true;
    }
    return false;
}

Key points:

• The retry uses a fixed 2,000 ms wait. Adjust based on your traffic pattern.

• The isChannelClosed method checks for channel.close in the error message, which covers both reply-code=530 (throttling) and other channel closure scenarios.

• After recovery, resume operations on the new channel object.