ApsaraMQ for RocketMQ:Envoyer et recevoir des messages avec le SDK Java

Push consumer

Créez un fichier PushConsumerExample.java et exécutez-le :

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;

public class PushConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(PushConsumerExample.class);

    private PushConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException, InterruptedException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // Uncomment for serverless instances accessed over the Internet:
                // .setNamespace("<your-instance-id>")
                // Uncomment for Internet access or serverless VPC without authentication-free:
                // .setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"))
                .build();

        // Subscribe to all messages in the topic (tag = "*")
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .setMessageListener(messageView -> {
                    System.out.println("Consume Message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();

        // Keep the consumer running
        Thread.sleep(Long.MAX_VALUE);

        // To shut down the consumer gracefully, call:
        // pushConsumer.close();
    }
}

Pour une logique métier longue, privilégiez un simple consumer (mode extraction). Un simple consumer vous permet de contrôler exactement le nombre de messages extraits à la fois ; vous n'extrayez le lot suivant qu'après avoir traité et accusé réception du lot actuel. Vous pouvez définir InvisibleDuration (la période pendant laquelle un message reçu reste invisible pour les autres consommateurs, de 10 secondes à 12 heures) pour contrôler le délai d'expiration de l'accusé de réception, et l'étendre dynamiquement si nécessaire.

Simple consumer

Créez un fichier SimpleConsumerExample.java et exécutez-le :

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.SimpleConsumer;
import org.apache.rocketmq.client.apis.message.MessageId;
import org.apache.rocketmq.client.apis.message.MessageView;
import org.apache.rocketmq.shaded.org.slf4j.Logger;
import org.apache.rocketmq.shaded.org.slf4j.LoggerFactory;

import java.io.IOException;
import java.time.Duration;
import java.util.Collections;
import java.util.List;

public class SimpleConsumerExample {
    private static final Logger LOGGER = LoggerFactory.getLogger(SimpleConsumerExample.class);

    private SimpleConsumerExample() {
    }

    public static void main(String[] args) throws ClientException, IOException {
        String endpoints = "<your-endpoint>";
        String topic = "<your-topic>";
        String consumerGroup = "<your-consumer-group>";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // Uncomment for serverless instances accessed over the Internet:
                // .setNamespace("<your-instance-id>")
                // Uncomment for Internet access or serverless VPC without authentication-free:
                // .setCredentialProvider(new StaticSessionCredentialsProvider("<instance-username>", "<instance-password>"))
                .build();

        // Long polling timeout
        Duration awaitDuration = Duration.ofSeconds(10);
        // Subscribe to all messages in the topic (tag = "*")
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);

        SimpleConsumer consumer = provider.newSimpleConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setConsumerGroup(consumerGroup)
                .setAwaitDuration(awaitDuration)
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                .build();

        // Maximum messages per pull
        int maxMessageNum = 16;
        // Messages stay invisible to other consumers for this duration after being received
        Duration invisibleDuration = Duration.ofSeconds(10);

        // Poll for messages in a loop. For real-time consumption, use multiple threads.
        while (true) {
            final List<MessageView> messages = consumer.receive(maxMessageNum, invisibleDuration);
            messages.forEach(messageView -> {
                System.out.println("Received message: " + messageView);
            });
            for (MessageView message : messages) {
                final MessageId messageId = message.getMessageId();
                try {
                    // ACK each message to commit the consumption result to the broker
                    consumer.ack(message);
                    System.out.println("Message is acknowledged successfully, messageId= " + messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
        // To shut down the consumer gracefully, call:
        // consumer.close();
    }
}

SDK for Java 5.x (rocketmq-client-java)

Version minimale : 5.0.6

Définissez le namespace dans ClientConfiguration :

ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
    .setEndpoints(endpoints)
    .setNamespace("InstanceId")
    .setCredentialProvider(sessionCredentialsProvider)
    .build();

SDK for Java 5.x (rocketmq-client)

Version minimale : 5.2.0

Définissez le namespace séparément sur le producteur et le consommateur :

// Producer
producer.setNamespaceV2("InstanceId");

// Consumer
consumer.setNamespaceV2("InstanceId");

TCP client SDK for Java 1.x

Version minimale : 1.9.0.Final

Définissez le namespace via les propriétés :

properties.setProperty(PropertyKeyConst.Namespace, "InstanceId");