ApsaraMQ for RabbitMQ:Notes d'utilisation pour les clients RabbitMQ open source

Activer la récupération automatique de connexion

Activez la récupération automatique de connexion sur com.rabbitmq.client.ConnectionFactory. Sans cette option, les mises à niveau du broker interrompent les lectures et écritures de messages, et le client ne peut pas se reconnecter automatiquement.

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

Utiliser des connexions persistantes

N'ouvrez et ne fermez pas de connexions pour chaque opération de publication ou de consommation. Les reconnections fréquentes gaspillent les ressources réseau et du broker, et peuvent déclencher la protection contre les inondations SYN.

Pour plus d'informations, consultez la section Connections and channels.

Éviter les opérations de métadonnées dans le code applicatif

Les opérations de métadonnées telles que queueDeclare et exchangeDeclare sont soumises à une limitation de débit. Si elles sont déclenchées, le broker peut fermer la connexion.

Créez les files d'attente et les exchanges via la console ApsaraMQ for RabbitMQ au lieu de les déclarer au moment de l'exécution lors de l'envoi de messages. Pour plus d'informations sur les limites de limitation de débit, consultez la section Limits.

Configurer les confirmations de l'éditeur

Les confirmations de l'éditeur permettent au broker d'accuser réception de chaque message après son écriture sur le disque, afin que le client puisse détecter et retenter les livraisons échouées. L'activation des confirmations de l'éditeur fait passer le mode d'envoi de synchrone à asynchrone, ce qui peut augmenter la latence car le broker renvoie une réponse de succès uniquement après avoir confirmé la persistance du message.

Pour activer les confirmations de l'éditeur sur un canal :

1. Appelez confirmSelect() lors de la création du canal.

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

2. Après avoir appelé basicPublish, attendez la réponse du broker et traitez le résultat.

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

Gérer les messages non routables avec l'indicateur mandatory

La définition du paramètre mandatory sur true amène le broker à invoquer le rappel ReturnListener lorsqu'un message ne peut être routé vers aucune file d'attente.

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

Attribuer un ID de message à chaque message

Attribuez un ID personnalisé à chaque message afin de pouvoir interroger sa trace et résoudre les problèmes de livraison. Pour plus d'informations, consultez la section How to set a message ID

Gérer les erreurs de publication

Déterminez la stratégie de gestion des erreurs en fonction du type d'erreur renvoyé par basicPublish :

• Exceptions métier (telles que ExchangeNotExist) : Levez l'exception immédiatement.

• Limitation de débit ou fermeture de canal : Fermez la connexion existante, créez et initialisez un nouveau canal, puis retentez l'envoi.

Exemple :

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

Pour l'exemple de code complet, consultez la section Produce messages.

Configurer le mode de connexion pour les clients Spring Boot

Lors de l'utilisation d'un client Spring Boot, vous devez définir explicitement le mode de cache de connexion sur connection plutôt que de vous fier au mode channel par défaut. Le même principe s'applique aux autres bibliothèques clientes qui prennent en charge la mise en cache des connexions : configurez toujours explicitement le mode de connexion.

Par exemple, définissez la propriété suivante dans application.properties ou application.yml :

spring.rabbitmq.cache.connection.mode=connection

Par exemple, définissez la propriété suivante dans application.properties ou application.yml :

spring.rabbitmq.cache.connection.mode=connection

Établir plusieurs connexions par producteur

Lors de la production de messages, chaque client producteur doit établir plusieurs connexions pour se connecter à plusieurs nœuds de service backend. Cela permet d'éviter les goulots d'étranglement des ressources causés par des pics de trafic soudains sur une seule connexion.

Recommandation :

• Chaque client producteur doit établir 50 connexions vers le backend.

• Ces connexions doivent être réparties sur plusieurs nœuds de service backend.

Prévenir le déséquilibre de consommation

Assurez-vous que les messages sont répartis uniformément entre les consommateurs. Un déséquilibre de consommation entraîne une surcharge de certains consommateurs tandis que d'autres restent inactifs. Pour les détails de configuration, consultez la section « Notes d'utilisation » de la rubrique Connections and channels.

Utiliser les tags de consommateur générés par le broker

Laissez le broker générer un tag de consommateur globalement unique plutôt que d'en spécifier un manuellement. Si vous spécifiez un tag de consommateur personnalisé, assurez-vous qu'il est unique parmi tous les consommateurs.

Configurer la QoS (prefetch)

Appelez basicQos pour limiter le nombre de messages non acquittés que le broker pousse vers un consommateur. Lorsque cette limite est atteinte, le broker suspend la livraison jusqu'à ce que le consommateur acquitte les messages en attente.

Options de portée :

Contraintes de la plateforme :

• La valeur QoS par défaut est de 100 par consommateur, équivalente à channel.basicQos(100, false).

• Les valeurs QoS personnalisées ne peuvent pas dépasser 100. Les valeurs supérieures à 100 sont ignorées et la valeur par défaut est utilisée à la place.

• basicQos n'a aucun effet en mode autoACK.

Choix d'une valeur QoS :

Délais d'attente ACK et comportement de nouvelle tentative

Si un consommateur n'acquitte pas un message dans le délai configuré, le broker le relivre.

Chaque message peut faire l'objet d'un nombre maximal de nouvelles tentatives configuré. Une fois toutes les tentatives épuisées, le message est supprimé ou routé vers l'exchange de lettres mortes.

Le délai d'attente et les limites de nouvelles tentatives dépendent du type d'instance et de l'édition :

Utiliser basicConsume au lieu de basicGet

basicGet interroge les messages individuellement et présente des limites de débit et de transactions par seconde (TPS) inférieures à celles de basicConsume. Dans les environnements de production avec des volumes de messages élevés, utilisez basicConsume avec un consommateur de longue durée pour une livraison basée sur le push.