ApsaraMQ for RabbitMQ:Intégration Spring

# The endpoint. Obtain the endpoint on the Instance Details page in the ApsaraMQ for RabbitMQ console. 
spring.rabbitmq.host=XXXXXX.amqp.aliyuncs.com
# The port used to connect to ApsaraMQ for RabbitMQ.
spring.rabbitmq.port=5672
# The static username of the instance. View the username on the Static Accounts page in the ApsaraMQ for RabbitMQ console. 
spring.rabbitmq.username=******
# The static password of the instance. View the password on the Static Accounts page in the ApsaraMQ for RabbitMQ console. 
spring.rabbitmq.password=******
# The virtual host, which provides logical isolation. View the virtual host on the Vhosts page in the ApsaraMQ for RabbitMQ console. 
spring.rabbitmq.virtual-host=test_vhost
# The message acknowledgment (Ack) mode.
# 1. none: After a consumer receives a message, the server considers the message successfully processed, regardless of whether the consumption is successful. This is the autoAck mode in RabbitMQ.
# 2. auto: The client automatically sends an ack after a message is successfully consumed. If the message fails to be processed, the client sends a nack or throws an exception. You do not need to explicitly call Channel.basicAck().
# 3. manual: Manually send an Ack. You must explicitly call Channel.basicAck() after a message is successfully consumed.
spring.rabbitmq.listener.simple.acknowledge-mode=manual

# Set the cache mode to CONNECTION. ApsaraMQ for RabbitMQ uses a distributed architecture. In CONNECTION mode, clients can connect to multiple service nodes in the cluster in a more balanced way. This method can effectively prevent load hot spots and improve the efficiency of message sending and consumption.
# Note: In CONNECTION mode, automatic declaration by RabbitAdmin (auto-creating Exchanges, Queues, and Bindings) does not take effect. You need to manually declare the required topology resources.
spring.rabbitmq.cache.connection.mode=connection
# Adjust the value as needed.
spring.rabbitmq.cache.connection.size=50
# Adjust the value as needed.
spring.rabbitmq.cache.channel.size=1

# The maximum number of unacknowledged (Ack) messages that a consumer can process at a time (QoS). The ApsaraMQ for RabbitMQ server uses min{prefetch, 100} as the QoS value. If the consumer has low processing capabilities, decrease this value.
spring.rabbitmq.listener.simple.prefetch=100
# The minimum number of concurrent consumers for the RabbitMQ listener. Adjust the value as needed.
spring.rabbitmq.listener.simple.concurrency=10
# The maximum number of concurrent consumers for the RabbitMQ listener. When the consumption rate is high enough, the client starts max-concurrency consumers to consume messages.
spring.rabbitmq.listener.simple.max-concurrency=20

Bonnes pratiques pour les connexions et les canaux

ApsaraMQ for RabbitMQ utilise une architecture distribuée multi-nœuds en backend. Pour éviter les goulots d'étranglement de performance au niveau des nœuds uniques côté serveur, suivez ces bonnes pratiques lors de la configuration des connexions client.

• Utilisez le mode CONNECTION : Configurez le client pour établir plusieurs connexions TCP vers le serveur. En mode CONNECTION, un seul client répartit les connexions sur plusieurs nœuds de service backend, exploitant ainsi les ressources de plusieurs nœuds du cluster. Nous vous recommandons de configurer le nombre de connexions (CONNECTION) par client à une dizaine, en ajustant selon le volume de trafic du client.

• Évitez l'établissement et la fermeture fréquents des connexions : Ne créez et ne fermez pas fréquemment les connexions. Le cycle fréquent des connexions réduit l'efficacité de l'envoi et de la consommation des messages. Maintenez des connexions persistantes plutôt que d'en créer une par opération.

• Limitez le nombre de canaux par connexion : Définissez le nombre de canaux pouvant être créés sur une seule connexion à 50 maximum. Un trop grand nombre de canaux sur une seule connexion peut provoquer des pics soudains de trafic TCP. Combiné à un réordonnancement occasionnel des paquets, cela peut déclencher le mécanisme Linux Netfilter Conntrack pour classer les paquets comme INVALID, ce qui amène le client à envoyer activement un RST pour réinitialiser la connexion.

Configurez le mode de connexion et le nombre de connexions/canaux à l'aide des paramètres suivants :

• spring.rabbitmq.cache.connection.mode - Définissez sur connection pour le mode CONNECTION.

• spring.rabbitmq.cache.connection.size - Nombre de connexions. Recommandé : une dizaine.

• spring.rabbitmq.cache.channel.size - Nombre de canaux par connexion. Recommandé : 50 maximum.

Produire des messages

Dans RabbitMQService, obtenez RabbitTemplate via l'injection de dépendances et appelez sa méthode send pour envoyer un message.

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

Consommer des messages

Utilisez l'annotation @RabbitListener pour consommer des messages :

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

Voici les configurations optionnelles courantes pour RabbitListener :

Remarques importantes

Vérifiez si votre client utilise la méthode receiveAndConvert de RabbitTemplate pour la consommation des messages.

Cette méthode accepte des paramètres tels que le nom de la file d'attente et la durée du délai d'expiration. Lorsqu'un délai d'expiration est spécifié, la logique de traitement interne présente le problème suivant : avant de consommer un message, la méthode envoie d'abord une commande BasicCancel pour annuler l'abonnement du consommateur, puis envoie un Ack pour confirmer le message.

Cependant, lorsqu'ApsaraMQ for RabbitMQ reçoit une demande BasicCancel, il annule immédiatement le consommateur correspondant et remet en file d'attente tous les messages non accusés de réception. Cela signifie que le Ack envoyé ultérieurement par le client est effectivement invalide — le serveur ne le traite pas comme un accusé de réception valide mais rediffuse ces messages, provoquant une consommation en double côté client.

De plus, si la file d'attente a un débit de messages élevé, les opérations fréquentes d'annulation et de remise en file d'attente peuvent entraîner une accumulation de messages, affectant davantage la stabilité de la consommation.