ApsaraMQ for RabbitMQ:Notas de uso para clientes RabbitMQ de código aberto

Ative a recuperação automática de conexão

Ative a recuperação automática de conexão em com.rabbitmq.client.ConnectionFactory. Sem essa configuração, as atualizações do broker interrompem a leitura e a gravação de mensagens, e o cliente não consegue se reconectar sozinho.

// Enable automatic connection recovery
factory.setAutomaticRecoveryEnabled(true);
// Set the interval between recovery attempts (milliseconds)
factory.setNetworkRecoveryInterval(5000);

Use conexões persistentes

Não abra e feche conexões a cada operação de publicação ou consumo. Reconexões frequentes desperdiçam recursos de rede e do broker, e podem acionar a proteção contra SYN flood.

Para mais informações, consulte Connections and channels.

Evite operações de metadados no código da aplicação

Operações de metadados, como queueDeclare e exchangeDeclare, estão sujeitas a limitação de taxa. Se isso ocorrer, o broker poderá fechar a conexão.

Crie filas e exchanges pelo console do ApsaraMQ for RabbitMQ em vez de declará-los em tempo de execução ao enviar mensagens. Para mais informações sobre os limites de taxa, consulte Limits.

Configure as confirmações de publicação

As confirmações de publicação permitem que o broker confirme cada mensagem após gravá-la em disco, para que o cliente detecte e repita entregas com falha. Ativar as confirmações de publicação altera o modo de envio de síncrono para assíncrono. Isso pode aumentar a latência, pois o broker retorna uma resposta de sucesso apenas após confirmar a persistência da mensagem.

Para ativar as confirmações de publicação em um canal:

1. Chame confirmSelect() ao criar o canal.

   // Create a connection and a channel
   Connection connection = createConnection(hostName, userName, passWord, virtualHost);
   Channel channel = connection.createChannel();
   // Enable publisher confirms
   channel.confirmSelect();

2. Após chamar basicPublish, aguarde a resposta do broker e trate o resultado.

   // Publish a message (reuse the existing channel)
   channel.basicPublish(exchangeName, bindingKey, true, props,
                   ("example body").getBytes(StandardCharsets.UTF_8));
   
   // Wait up to 3 seconds for the confirm
   if (channel.waitForConfirms(3000)) {
       // Message persisted to disk successfully
   } else {
       // Confirm failed -- resend the message
   }

Trate mensagens não roteáveis com a flag mandatory

Definir o parâmetro mandatory como true leva o broker a invocar o callback ReturnListener quando não é possível rotear uma mensagem para nenhuma fila.

// Register a callback for unroutable messages
channel.addReturnListener(returnMessage ->
    System.out.println("return msgId=" + returnMessage.getProperties().getMessageId()));
// Publish with mandatory=true (third parameter)
channel.basicPublish(exchangeName, routingKey, true, props,
    content.getBytes(StandardCharsets.UTF_8));

Atribua um ID de mensagem a cada mensagem

Atribua um ID personalizado a cada mensagem para consultar o rastreamento e solucionar problemas de entrega. Para mais informações, consulte How to set a message ID

Trate erros de publicação

Determine a estratégia de tratamento de erros com base no tipo de erro retornado por basicPublish:

• Exceções de negócio (como ExchangeNotExist): Lance a exceção imediatamente.

• Limitação de taxa ou fechamento de canal: Feche a conexão existente, crie e inicialize um novo canal e repita o envio.

Exemplo:

private void doSend(String content) throws Exception {
    try {
        // Assign a unique message ID
        String msgId = UUID.randomUUID().toString();
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().messageId(msgId).build();

        channel.basicPublish(exchangeName, routingKey, true, props, content.getBytes(StandardCharsets.UTF_8));
        // Wait up to 3 seconds for the confirm
        if (channel.waitForConfirms(3000)) {
            // Message persisted to disk successfully
        } else {
            // Confirm failed -- resend the message
        }

    } catch (Exception e) {
        String message = e.getMessage();
        System.out.println(message);

        if (channelClosedByServer(message)) {
            // Recover: close the old channel, create and reinitialize a new one, then retry
            factory.closeCon(channel);
            channel = factory.createChannel();
            this.initChannel();
            doSend(content);
        } else {
            // Non-broker error -- propagate the exception
            throw e;
        }
    }
}

private boolean channelClosedByServer(String errorMsg) {
    if (errorMsg != null && errorMsg.contains("channel.close")) {
        return true;
    } else {
        return false;
    }
}

Para o código de exemplo completo, consulte Produce messages.

Configure o modo de conexão para clientes Spring Boot

Ao usar um cliente Spring Boot, defina explicitamente o modo de cache de conexão como connection em vez de depender do modo channel padrão. Esse mesmo princípio se aplica a outras bibliotecas de cliente que suportam cache de conexão: sempre configure o modo de conexão explicitamente.

Por exemplo, defina a seguinte propriedade em application.properties ou application.yml:

spring.rabbitmq.cache.connection.mode=connection

Por exemplo, defina a seguinte propriedade em application.properties ou application.yml:

spring.rabbitmq.cache.connection.mode=connection

Estabeleça múltiplas conexões por produtor

Ao produzir mensagens, cada cliente produtor deve estabelecer múltiplas conexões para se conectar a vários nós de service de backend. Isso evita que picos repentinos de tráfego em uma única conexão causem gargalos de recursos.

Recomendação:

• Cada cliente produtor deve estabelecer 50 conexões com o backend.

• Distribua essas conexões entre vários nós de service de backend.

Evite a assimetria de consumo

Garanta a distribuição uniforme das mensagens entre os consumidores. A assimetria de consumo sobrecarrega alguns consumidores enquanto outros permanecem ociosos. Para obter detalhes de configuração, consulte a seção "Notas de uso" em Connections and channels.

Use tags de consumidor geradas pelo broker

Permita que o broker gere uma tag de consumidor globalmente única em vez de especificá-la manualmente. Se especificar uma tag de consumidor personalizada, certifique-se de que seja única entre todos os consumidores.

Configure o QoS (prefetch)

Chame basicQos para limitar o número de mensagens não confirmadas que o broker envia a um consumidor. Ao atingir esse limite, o broker pausa a entrega até que o consumidor confirme as mensagens pendentes.

Opções de escopo:

Restrições da plataforma:

• O valor padrão de QoS é 100 por consumidor, equivalente a channel.basicQos(100, false).

• Os valores personalizados de QoS não podem exceder 100. O broker ignora valores acima de 100 e aplica o padrão.

• O basicQos não tem efeito no modo autoACK.

Como escolher um valor de QoS:

Timeouts de ACK e comportamento de nova tentativa

Se um consumidor não confirmar uma mensagem dentro do timeout configurado, o broker a entrega novamente.

O broker pode tentar entregar cada mensagem até o máximo configurado. Após todas as tentativas se esgotarem, o broker descarta a mensagem ou a roteia para o exchange de dead-letter.

Os limites de timeout e de tentativas dependem do tipo e da edição da instância:

Use basicConsume em vez de basicGet

O basicGet faz polling de mensagens individuais e possui limites de throughput e transações por segundo (TPS) mais baixos em comparação com o basicConsume. Em ambientes de produção com grandes volumes de mensagens, use o basicConsume com um consumidor de longa duração para entrega baseada em push.