ApsaraMQ for RabbitMQ:Best practices for instance throttling

TPS total da instância

Escolha a abordagem adequada ao seu padrão de tráfego:

TPS de SendMessage por nó

O ApsaraMQ for RabbitMQ distribui o tráfego entre vários nós de backend. Se o tráfego se concentrar em um único nó, o limite por nó será acionado mesmo que haja capacidade disponível no nível da instância.

Para distribuir a carga uniformemente entre os nós:

• Abra pelo menos 10 conexões por fila. Cada conexão pode ser roteada para um nó diferente e distribuir a carga de trabalho de envio.

• Usuários do Spring: defina o CachingConnectionFactory para o modo CONNECTION. Essa configuração cria uma nova conexão por sessão em vez de multiplexar canais em uma única conexão. Consulte Spring integration.

CloudMonitor (recomendado)

• Granularidade: nível de minuto (período de 1 minuto).

• Escopo: pico de TPS no nível da instância.

• Custo: gratuito.

• Exibe tendências de TPS dos últimos 14 dias.

• Permite configurar regras de alerta para notificar antes da ocorrência de throttling.

Consultar o pico de TPS de uma instância e configurar uma regra de alerta usando o CloudMonitor.

Página Instance Details (recomendado)

• Granularidade: pico de TPS no nível de segundo.

• Escopo: pico de TPS no nível da instância e por operação de API.

• Custo: gratuito.

• Identifica picos curtos de tráfego com precisão de segundos.

• Permite filtragem por operação de API.

Apenas os primeiros 10 minutos de resultados são exibidos.

Consultar o pico de TPS de uma instância na página Instance Details.

Simple Log Service

• Granularidade: pico de TPS no nível de segundo.

• Escopo: pico de TPS no nível da instância.

• Custo: cobrado pelo Simple Log Service (consulte Itens faturáveis do pagamento por recurso).

• Suporta instruções de consulta SLS para análises avançadas.

Requer familiaridade com a sintaxe de consulta SLS. Os resultados podem ser mais difíceis de interpretar do que no CloudMonitor ou na página Instance Details.

Consultar o pico de TPS de uma instância usando o Simple Log Service.

TPS total da instância

TPS de SendMessage por nó

O broker limita o TPS de SendMessage em cada nó de backend da instância.

Limites por operação de API

Esses limites aplicam-se por instância. Instâncias serverless com cluster exclusivo não possuem throttling por API.

Código e mensagem de erro

Quando o throttling é acionado, o broker fecha o canal afetado e retorna:

• Código de erro: reply-code=530

• Mensagem de erro: 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)

O erro inclui o ID da requisição (ReqId), a fila de destino (dstQueue), a exchange de origem (srcExchange) e a chave de binding (bindingKey).

Recuperação após fechamento de canal

Como apenas o canal é fechado, capture a exceção AlreadyClosedException e recrie-o. O exemplo Java a seguir demonstra a recuperação baseada em tentativas:

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;
}

Pontos principais:

• A nova tentativa utiliza uma espera fixa de 2.000 ms. Ajuste esse valor conforme o seu padrão de tráfego.

• O método isChannelClosed verifica a presença de channel.close na mensagem de erro. Isso abrange tanto reply-code=530 (throttling) quanto outros cenários de fechamento de canal.

• Após a recuperação, retome as operações no novo objeto de canal.