ApsaraMQ for RocketMQ:Consumer types

Fonctionnement

Le SDK utilise en interne un modèle de thread Reactor :

1. Un thread d'interrogation longue intégré extrait les messages du serveur de manière asynchrone.

2. Les messages sont placés dans une file d'attente de cache interne.

3. Le SDK distribue les messages aux threads de consommation, qui invoquent votre écouteur.

Résultats de l'écouteur

L'écouteur de messages doit renvoyer l'un des résultats suivants :

Contraintes de fiabilité

Le Push consumer détermine le succès ou l'échec strictement à partir de la valeur de retour de l'écouteur. Pour préserver cette garantie :

1. Traitez de manière synchrone. Terminez tout le traitement avant de renvoyer le résultat.

2. Ne redistribuez pas les messages. Ne confiez pas les messages à d'autres threads et ne renvoyez pas de résultat avant que ces threads n'aient terminé.

Si l'écouteur renvoie un succès avant la fin du traitement et que ce dernier échoue ensuite, le serveur considère le message comme consommé et n'effectue pas de nouvelle tentative.

Livraison des messages ordonnés

Lorsqu'un groupe de consommateurs utilise le mode de consommation ordonnée, le Push consumer invoque l'écouteur dans l'ordre strict des messages sans configuration supplémentaire. Pour plus d'informations, consultez Messages ordonnés.

Quand utiliser les Push consumers

Les Push consumers sont idéaux lorsque :

• Le temps de traitement est prévisible. Des durées imprévisibles déclenchent de fréquents dépassements de délai, entraînant des messages en double via des nouvelles tentatives.

• Une consommation standard suffit. Le SDK contrôle le modèle de thread et livre les messages avec un débit maximal. Cela simplifie le développement mais ne prend pas en charge le traitement asynchrone ni le contrôle de débit personnalisé.

Classes SDK

ApsaraMQ for RocketMQ fournit deux classes SDK pour les Push consumers :

• **PushConsumer** -- Consomme des messages provenant de topics standards (non-Lite).

• **LitePushConsumer** -- Consomme des messages provenant de topics de type Lite, avec un contrôle de la consommation à la granularité du topic Lite.

// Consume normal messages with PushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

PushConsumer pushConsumer = provider.newPushConsumerBuilder()
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind the subscription
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        // Register the message listener
        .setMessageListener(new MessageListener() {
            @Override
            public ConsumeResult consume(MessageView messageView) {
                // Process the message and return the result
                return ConsumeResult.SUCCESS;
            }
        })
        .build();

// Consume normal messages with LitePushConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();

LitePushConsumer litePushConsumer = provider.newLitePushConsumerBuilder()
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Set the topic
        .bindTopic("<your-topic>")
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Register the message listener
        .setMessageListener(messageView -> {
            // Process the message and return the result
            return ConsumeResult.SUCCESS;
        })
        .build();

// Subscribe to Lite topics
litePushConsumer.subscribeLite("<your-lite-topic-1>");
litePushConsumer.subscribeLite("<your-lite-topic-2>");

Fonctionnement

1. Appelez ReceiveMessage pour extraire un lot de messages du serveur.

2. Distribuez les messages à vos threads métier pour traitement.

3. Appelez AckMessage pour chaque message traité avec succès.

En cas d'échec du traitement, n'envoyez pas d'accusé de réception. Le message redevient disponible après l'expiration de la durée d'invisibilité, ce qui déclenche une nouvelle tentative. Pour plus d'informations, consultez Politique de nouvelle tentative de SimpleConsumer.

Opérations API

Gestion des échecs

Le tableau suivant décrit l'impact des différents scénarios d'échec sur la livraison des messages :

Livraison des messages ordonnés

Un Simple consumer traite les messages ordonnés selon l'ordre de stockage. Pour un groupe de messages devant rester séquentiels, le message suivant ne peut être récupéré tant que le précédent n'est pas traité.

Quand utiliser les Simple consumers

Les Simple consumers sont idéaux lorsque :

• Le temps de traitement est imprévisible. Spécifiez une durée d'invisibilité initiale du message lors de l'appel à ReceiveMessage, puis prolongez-la avec ChangeInvisibleDuration si nécessaire.

• Des workflows personnalisés sont requis. Le SDK n'impose aucun modèle de threading -- implémentez une distribution asynchrone, une consommation par lots ou tout autre modèle personnalisé.

• Le contrôle du débit est important. Votre code décide quand et à quelle fréquence appeler ReceiveMessage, offrant un contrôle direct sur le débit.

Classe SDK

ApsaraMQ for RocketMQ fournit une classe SDK pour les Simple consumers : SimpleConsumer. Cette classe ne peut pas consommer de messages provenant de topics Lite.

// Consume normal messages with SimpleConsumer
ClientServiceProvider provider = ClientServiceProvider.loadService();
String topic = "<your-topic>";
FilterExpression filterExpression = new FilterExpression("<your-filter-tag>", FilterExpressionType.TAG);

SimpleConsumer simpleConsumer = provider.newSimpleConsumerBuilder()
        // Set the consumer group
        .setConsumerGroup("<your-consumer-group>")
        // Set the access endpoint
        .setClientConfiguration(ClientConfiguration.newBuilder().setEndpoints("<your-endpoint>").build())
        // Bind the subscription
        .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
        .build();

try {
    // Pull up to 10 messages, wait up to 30 seconds
    List<MessageView> messageViewList = simpleConsumer.receive(10, Duration.ofSeconds(30));
    messageViewList.forEach(messageView -> {
        System.out.println(messageView);
        // Acknowledge each message after successful processing
        try {
            simpleConsumer.ack(messageView);
        } catch (ClientException e) {
            e.printStackTrace();
        }
    });
} catch (ClientException e) {
    // Handle failures such as throttling, then retry the receive call
    e.printStackTrace();
}

Contrôle du temps de traitement pour les Push consumers

Maintenez le traitement des messages dans les limites du seuil de délai. Des dépassements de délai fréquents provoquent des nouvelles tentatives inutiles et des messages en double. Si votre application gère régulièrement des tâches de longue durée, basculez vers un Simple consumer et définissez une durée d'invisibilité des messages appropriée.