ApsaraMQ for RocketMQ:Consumer load balancing

Fonctionnement

Le broker répartit uniformément les messages individuels d'une rubrique entre tous les consommateurs d'un groupe, indépendamment de la file d'attente à laquelle ils appartiennent. Plusieurs consommateurs peuvent traiter simultanément les messages d'une même file d'attente.

Dans cet exemple, le groupe de consommateurs A comprend trois consommateurs : A1, A2 et A3. Tous trois consomment les messages de Queue1. Le broker attribue chaque message à un seul consommateur à la fois.

Dès qu'un consommateur reçoit un message, le broker verrouille ce dernier, le rendant invisible aux autres consommateurs. Le verrouillage persiste jusqu'à ce que le consommateur accuse réception du message ou que le délai d'expiration du verrou soit atteint. Ce mécanisme empêche tout double traitement dans des conditions normales.

Gestion des messages ordonnés

Pour les messages ordonnés, ApsaraMQ for RocketMQ garantit que les messages appartenant au même groupe sont traités dans l'ordre exact de leur stockage sur le broker.

Prenons quatre messages ordonnés (M1 à M4) dans le groupe de messages G1 de Queue1. Si le consommateur A1 traite M1 et M2, le consommateur A2 ne peut commencer à traiter M3 ou M4 qu'une fois qu'A1 a soumis l'état de consommation de M1 et M2. Le broker applique un verrouillage séquentiel strict afin de préserver l'ordre.

Avantages par rapport à l'équilibrage de charge basé sur les files d'attente

Quand l'utiliser

L'équilibrage de charge basé sur les messages convient à la plupart des charges de travail de traitement d'événements en ligne où chaque message est géré indépendamment, comme le traitement des commandes, la diffusion de notifications ou la gestion d'événements en temps réel.

Pour les charges de travail de traitement de flux ou d'agrégation nécessitant le regroupement par lot des messages issus d'une même file d'attente, privilégiez plutôt l'équilibrage de charge basé sur les files d'attente.

Portée

L'équilibrage de charge basé sur les messages constitue la seule politique disponible pour les types Push Consumer et Simple Consumer sur la version 5.x du broker. Il est activé par défaut et ne requiert aucune configuration supplémentaire.

Exemple

Les types Push Consumer et Simple Consumer utilisent automatiquement l'équilibrage de charge basé sur les messages. Les exemples Java suivants illustrent ces deux types de consommateurs traitant des messages sans aucune configuration d'équilibrage de charge.

Push Consumer avec un écouteur de messages :

// Push Consumer: implement a message listener to process messages.
// Load balancing is handled automatically by the broker.
MessageListener messageListener = new MessageListener() {
    @Override
    public ConsumeResult consume(MessageView messageView) {
        System.out.println(messageView);
        // Return the consumption result.
        return ConsumeResult.SUCCESS;
    }
};

Simple Consumer avec accusé de réception manuel :

// Simple Consumer: pull messages, process them, and acknowledge each one.
// Load balancing is handled automatically by the broker.
try {
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        try {
            // Acknowledge the message after processing.
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // If the pull fails due to throttling or other issues, retry the request.
    e.printStackTrace();
}

Fonctionnement

Le broker assigne chaque file d'attente d'une rubrique à exactement un consommateur du groupe. Chaque consommateur traite alors l'ensemble des messages provenant des files d'attente qui lui sont attribuées.

Dans cet exemple, une rubrique comporte trois files d'attente (Queue1, Queue2, Queue3) et le groupe de consommateurs deux consommateurs. Comme chaque file d'attente est attribuée à un seul consommateur, le consommateur A2 obtient deux files d'attente tandis qu'A1 n'en reçoit qu'une. S'il y a moins de files d'attente que de consommateurs, certains consommateurs ne reçoivent aucune file d'attente et restent inactifs.

Chaque consommateur suit cette séquence de traitement : extraction des messages depuis la file d'attente attribuée, soumission de l'offset de consommation et persistance de cet offset. Puisque l'état de consommation n'est pas renvoyé à la file d'attente lorsque le consommateur extrait les messages, chaque file d'attente doit être exclusivement assignée à un seul consommateur afin d'éviter tout double traitement.

Avantages par rapport à l'équilibrage de charge basé sur les messages

Quand l'utiliser

L'équilibrage de charge basé sur les files d'attente convient particulièrement aux applications de calcul de flux et d'agrégation de données qui nécessitent le traitement par lot ou l'agrégation des messages provenant d'une même source. Par exemple, un consommateur peut collecter des métriques depuis une seule file d'attente sur une fenêtre temporelle donnée et calculer des moyennes continues.

Portée

L'équilibrage de charge basé sur les files d'attente constitue la seule politique disponible pour les consommateurs sur les versions 3.x et 4.x du broker, incluant les types Pull Consumer, Push Consumer par défaut, Pull Consumer par défaut et Lite Pull Consumer. Sur la version 5.x du broker, les types Pull Consumer continuent d'utiliser par défaut l'équilibrage de charge basé sur les files d'attente.

Aucune configuration supplémentaire n'est requise : l'équilibrage de charge basé sur les files d'attente est automatiquement activé pour ces types de consommateurs.

Exemple

Pour un exemple de code, consultez LitePullConsumerAssign.java dans la bibliothèque de code Apache RocketMQ.

Implémentez une gestion idempotente des messages

Les deux politiques d'équilibrage de charge déclenchent un rééquilibrage temporaire lors de l'ajout ou du retrait de consommateurs, ou lorsque le broker effectue une mise à l'échelle. Pendant le rééquilibrage, un petit nombre de messages peut être distribué plusieurs fois. Pour gérer cette situation, implémentez une déduplication afin de garantir l'idempotence de votre logique de consommation de messages.