ApsaraMQ for RabbitMQ:Spring integration

Melhores práticas para Connection e Channel

O ApsaraMQ for RabbitMQ utiliza uma arquitetura distribuída com múltiplos nós no backend. Para evitar gargalos de desempenho em um único nó no lado do servidor, siga estas melhores práticas ao configurar as conexões do cliente.

• Utilize o modo CONNECTION: Configure o cliente para estabelecer múltiplas conexões TCP com o servidor. Nesse modo, um único cliente distribui as conexões por vários nós de service do backend e aproveita os recursos de diversos nós do cluster. Recomendamos configurar o número de conexões (CONNECTION) por cliente na casa das dezenas e ajustar conforme o volume de tráfego.

• Evite criação e encerramento frequente de conexões: Não crie nem feche conexões constantemente. A ciclagem frequente de conexões reduz a eficiência no envio e consumo de mensagens. Mantenha conexões persistentes em vez de criá-las para cada operação.

• Limite o número de channels por conexão: Defina o número máximo de channels em uma única conexão como 50. Muitos channels em uma só conexão podem causar picos repentinos de tráfego TCP. Se combinado com reordenação ocasional de pacotes, esse cenário pode acionar o mecanismo Conntrack do Netfilter no Linux, classificar pacotes como INVALID e fazer com que o cliente envie ativamente um RST para redefinir a conexão.

Configure o modo de conexão e a quantidade de CONNECTION/CHANNEL pelos seguintes parâmetros:

• spring.rabbitmq.cache.connection.mode - Defina como connection para usar o modo CONNECTION.

• spring.rabbitmq.cache.connection.size - Número de conexões. Recomendado: na casa das dezenas.

• spring.rabbitmq.cache.channel.size - Número de channels por conexão. Recomendado: no máximo 50.

Produzir mensagens

Em RabbitMQService, obtenha o RabbitTemplate por injeção de dependência e chame seu método send para enviar uma mensagem.

import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.nio.charset.StandardCharsets;
import java.util.UUID;

@Service
public class RabbitMQService {
    
    @Autowired
    private RabbitTemplate rabbitTemplate;

    public void sendMessage(String exchange, String routingKey, String content) {
        // Set the MessageId.
        String msgId = UUID.randomUUID().toString();
        MessageProperties messageProperties = new MessageProperties();
        messageProperties.setMessageId(msgId);
        // Create a Message.
        Message message = new Message(content.getBytes(StandardCharsets.UTF_8), messageProperties);
        /*
         * Call the send() interface to send the message.
         * exchange: the name of the exchange.
         * routingKey: the routing key.
         * message: the message content.
         * correlationData is used for publisher confirms.
         */
        rabbitTemplate.send(exchange, routingKey, message, null);
    }
}

Consumir mensagens

Utilize a anotação @RabbitListener para consumir mensagens:

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.util.Arrays;

@Component
public class MessageListener {

     /**
     * Receive a message.
     * @param message The message.
     * @param channel The channel.
     * @throws IOException
     * Replace queues with the name of the queue that you created.
     */
    @RabbitListener(queues = "myQueue")
    public void receiveFromMyQueue(Message message, Channel channel) throws IOException {
        // Enter the business logic for message consumption.
        ...
        // You must return an Ack within the Ack validity period (consumption timeout). Otherwise, the confirmation is invalid and the message is redelivered.
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
    }
}

Veja a seguir as configurações opcionais comuns para RabbitListener:

Observações importantes

Verifique se o seu cliente utiliza o método receiveAndConvert do RabbitTemplate para consumo de mensagens.

Esse método aceita parâmetros como nome da fila e duração do tempo limite. Quando um tempo limite é especificado, a lógica interna de processamento apresenta o seguinte problema: antes de consumir uma mensagem, o método envia primeiro um comando BasicCancel para cancelar a assinatura do consumidor e, em seguida, envia um Ack para confirmar a mensagem.

No entanto, quando o ApsaraMQ for RabbitMQ recebe uma solicitação BasicCancel, ele cancela imediatamente o consumidor correspondente e recoloca na fila todas as mensagens não confirmadas. Isso significa que o Ack subsequente enviado pelo cliente torna-se efetivamente inválido — o servidor não o trata como uma confirmação válida, mas sim reenvia essas mensagens, causando consumo duplicado no lado do cliente.

Além disso, se a fila tiver um alto throughput de mensagens, operações frequentes de cancelamento e reenfileiramento podem levar ao acúmulo de mensagens e impactar ainda mais a estabilidade do consumo.