How to use the SDK for Java to subscribe to messages - ApsaraMQ for RocketMQ

Abonnez-vous aux messages à l'aide du SDK client TCP Java pour ApsaraMQ for RocketMQ. Cette rubrique présente les modes d'abonnement, les modes de distribution et des exemples de code complets.

Modes d'abonnement

ApsaraMQ for RocketMQ prend en charge deux modes d'abonnement : le clustering (regroupement) et le broadcasting (diffusion).

Clustering (par défaut)

En mode clustering, tous les consommateurs d'un même groupe se partagent équitablement la charge des messages. Chaque message est distribué à un seul consommateur.

Par exemple, si une rubrique contient 9 messages et que le groupe de consommateurs compte 3 consommateurs, chaque consommateur reçoit 3 messages.

// Clustering is the default mode. To set it explicitly:
properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);

Broadcasting

En mode broadcasting, chaque consommateur du groupe reçoit une copie intégrale de chaque message.

Par exemple, si une rubrique contient 9 messages et que le groupe de consommateurs compte 3 consommateurs, chaque consommateur reçoit les 9 messages.

// Set the subscription mode to broadcasting.
properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

Remarque

Toutes les instances de consommateurs partageant le même ID de groupe doivent utiliser des abonnements cohérents.
Le mode broadcasting ne prend pas en charge l'envoi ou la réception de messages ordonnés, le suivi de la progression de la consommation ni la réinitialisation des offsets des consommateurs. Pour plus d'informations, consultez la section Consommation en clustering et consommation en broadcasting.

Modes de distribution

ApsaraMQ for RocketMQ prend en charge deux modes de distribution : push (poussée) et pull (extraction).

Mode	Description	Idéal pour
Push	Les messages sont poussés depuis ApsaraMQ for RocketMQ vers les consommateurs. Prend également en charge la consommation par lots.	La plupart des cas d'utilisation
Pull	Les consommateurs extraient les messages depuis ApsaraMQ for RocketMQ, offrant davantage d'options de réception et plus de liberté dans l'extraction des messages.	Les cas d'utilisation nécessitant un contrôle fin de la récupération des messages

Important

Les consommateurs pull nécessitent une instance Enterprise Platinum Edition.
Les consommateurs pull ne peuvent se connecter que via un Virtual Private Cloud (VPC).

Pour obtenir la liste complète des méthodes et des paramètres des consommateurs pull, consultez la section Méthodes et paramètres.

Exemples de code

Les exemples suivants illustrent la consommation push, la consommation push par lots et la consommation pull. Pour plus d'exemples, consultez le référentiel de code ApsaraMQ for RocketMQ sur GitHub.

Remplacez les espaces réservés suivants par vos valeurs réelles :

Espace réservé	Description	Où le trouver
`<your-group-id>`	ID du groupe de consommateurs	Console ApsaraMQ for RocketMQ
`<your-tcp-endpoint>`	Endpoint TCP	Section TCP Endpoint de la page Instance Details dans la console

Tous les exemples lisent les identifiants AccessKey à partir des variables d'environnement ALIBABA_CLOUD_ACCESS_KEY_ID et ALIBABA_CLOUD_ACCESS_KEY_SECRET.

Consommateur push

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.MessageListener;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;

import java.util.Properties;

public class ConsumerTest {
   public static void main(String[] args) {
       Properties properties = new Properties();
       // The consumer group ID created in the ApsaraMQ for RocketMQ console.
       properties.put(PropertyKeyConst.GROUP_ID, "<your-group-id>");
       // Read AccessKey credentials from environment variables.
       properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
       properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
       // The TCP endpoint from the Instance Details page in the console.
       properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
       // Clustering mode (default). Uncomment the line below to switch to broadcasting.
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.CLUSTERING);
       // properties.put(PropertyKeyConst.MessageModel, PropertyValueConst.BROADCASTING);

       Consumer consumer = ONSFactory.createConsumer(properties);
       // Subscribe to specific tags. Use "||" as a separator.
       consumer.subscribe("TopicTestMQ", "TagA||TagB", new MessageListener() {
           public Action consume(Message message, ConsumeContext context) {
               System.out.println("Receive: " + message);
               return Action.CommitMessage;
           }
       });

       // Subscribe to another topic with all tags.
       // To unsubscribe, remove the subscription code and restart the consumer.
       consumer.subscribe("TopicTestMQ-Other", "*", new MessageListener() {
           public Action consume(Message message, ConsumeContext context) {
               System.out.println("Receive: " + message);
               return Action.CommitMessage;
           }
       });

       consumer.start();
       System.out.println("Consumer Started");
   }
}

Consommateur push (consommation par lots)

Important

La consommation par lots nécessite le SDK client TCP Java version 1.8.7.3 ou ultérieure. Pour plus de détails sur les versions, consultez les notes de version.

Deux propriétés contrôlent le comportement des lots :

Propriété	Description	Valeurs valides	Par défaut
`ConsumeMessageBatchMaxSize`	Nombre maximal de messages par lot. Le SDK appelle le rappel lorsque le nombre mis en cache atteint cette valeur.	1 à 1024	32
`BatchConsumeMaxAwaitDurationInSeconds`	Temps d'attente maximal en secondes avant la livraison d'un lot partiel. Le SDK appelle le rappel lorsque cette durée est écoulée, quelle que soit la taille du lot.	0 à 450	0

import com.aliyun.openservices.ons.api.Action;
import com.aliyun.openservices.ons.api.ConsumeContext;
import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.batch.BatchConsumer;
import com.aliyun.openservices.ons.api.batch.BatchMessageListener;
import java.util.List;
import java.util.Properties;

import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.tcp.example.MqConfig;

public class SimpleBatchConsumer {

    public static void main(String[] args) {
        Properties consumerProperties = new Properties();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        consumerProperties.setProperty(PropertyKeyConst.GROUP_ID, MqConfig.GROUP_ID);
        // Read AccessKey credentials from environment variables.
        consumerProperties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        consumerProperties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint from the Instance Details page in the console.
        consumerProperties.setProperty(PropertyKeyConst.NAMESRV_ADDR, MqConfig.NAMESRV_ADDR);

        // Set the maximum batch size to 128 messages.
        consumerProperties.setProperty(PropertyKeyConst.ConsumeMessageBatchMaxSize, String.valueOf(128));
        // Set the maximum wait time between batches to 10 seconds.
        consumerProperties.setProperty(PropertyKeyConst.BatchConsumeMaxAwaitDurationInSeconds, String.valueOf(10));

        BatchConsumer batchConsumer = ONSFactory.createBatchConsumer(consumerProperties);
        batchConsumer.subscribe(MqConfig.TOPIC, MqConfig.TAG, new BatchMessageListener() {

             @Override
            public Action consume(final List<Message> messages, ConsumeContext context) {
                System.out.printf("Batch-size: %d\n", messages.size());
                // Process messages in batches.
                return Action.CommitMessage;
            }
        });
        // Start the batch consumer.
        batchConsumer.start();
        System.out.println("Consumer start success.");

        // Keep the process alive to continue consuming.
        try {
            Thread.sleep(200000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

Consommateur pull

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.PullConsumer;
import com.aliyun.openservices.ons.api.TopicPartition;
import java.util.List;
import java.util.Properties;
import java.util.Set;

public class PullConsumerClient {
    public static void main(String[] args){
        Properties properties = new Properties();
        // The consumer group ID created in the ApsaraMQ for RocketMQ console.
        properties.setProperty(PropertyKeyConst.GROUP_ID, "<your-group-id>");
        // Read AccessKey credentials from environment variables.
        properties.put(PropertyKeyConst.AccessKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        properties.put(PropertyKeyConst.SecretKey, System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));
        // The TCP endpoint from the Instance Details page in the console.
        properties.put(PropertyKeyConst.NAMESRV_ADDR, "<your-tcp-endpoint>");
        PullConsumer consumer = ONSFactory.createPullConsumer(properties);
        // Start the pull consumer.
        consumer.start();
        // Get all partitions for the topic.
        Set<TopicPartition> topicPartitions = consumer.topicPartitions("topic-xxx");
        // Assign partitions to this consumer.
        consumer.assign(topicPartitions);

        while (true) {
            // Poll for messages with a 3-second timeout.
            List<Message> messages = consumer.poll(3000);
            System.out.printf("Received message: %s %n", messages);
        }
    }
}

Pour plus d'informations sur les partitions et les offsets, consultez la section Termes.

Nouvelle tentative de distribution des messages

Si la consommation d'un message échoue ou expire, ApsaraMQ for RocketMQ redistribue automatiquement le message. Pour plus d'informations, consultez la section Nouvelle tentative de distribution des messages.