ApsaraMQ for RocketMQ:Scheduled and delayed messages

Envoyer un message planifié

import org.apache.rocketmq.client.apis.*;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;

public class ProducerExample {
    public static void main(String[] args) throws ClientException {
        // Instance endpoint. Find this on the Endpoints tab of the Instance Details
        // page in the ApsaraMQ for RocketMQ console.
        // For internal access from ECS in the same VPC, use the VPC endpoint.
        // For public access, use the public endpoint and enable Internet access
        // on the instance.
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";

        // Topic name. Create this topic in the console first, with MessageType
        // set to Delay.
        String topic = "Your Topic";

        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);

        // For public endpoint access, provide your instance username and password.
        // Find these on the Intelligent Authentication tab of the Access Control
        // page in the console.
        // For VPC access from ECS, credentials are obtained automatically.
        // For serverless instances with authentication-free in VPCs enabled,
        // credentials are not required for VPC access.
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));

        ClientConfiguration configuration = builder.build();
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();

        // Deliver the message 10 minutes from now
        long deliverTimeStamp = System.currentTimeMillis() + 10L * 60 * 1000;

        Message message = provider.newMessageBuilder()
                .setTopic("topic")
                .setKeys("messageKey")       // Message key for tracing
                .setTag("messageTag")        // Tag for consumer-side filtering
                .setDeliveryTimestamp(deliverTimeStamp)
                .setBody("messageBody".getBytes())
                .build();
        try {
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

Consommer avec un consommateur push

Un consommateur push reçoit les messages via un callback d'écouteur de messages. Le broker pousse les messages vers le consommateur dès qu'ils sont prêts.

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 {
        // Instance endpoint (see producer example for details)
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";

        // Create the topic and consumer group in the console before running
        // this example
        String topic = "Your Topic";
        String consumerGroup = "Your ConsumerGroup";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);

        // Uncomment for public endpoint access (see producer example for details)
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        ClientConfiguration clientConfiguration = builder.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();

        Thread.sleep(Long.MAX_VALUE);
        // Shut down when no longer needed
        //pushConsumer.close();
    }
}

Consommer avec un consommateur simple

Un consommateur simple extrait (pull) les messages du broker et accuse explicitement réception de chacun d'eux après traitement.

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 {
        // Instance endpoint (see producer example for details)
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";

        // Create the topic and consumer group in the console before running
        // this example
        String topic = "Your Topic";
        String consumerGroup = "Your ConsumerGroup";

        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);

        // Uncomment for public endpoint access (see producer example for details)
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        ClientConfiguration clientConfiguration = builder.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();

        int maxMessageNum = 16;
        Duration invisibleDuration = Duration.ofSeconds(10);

        // Pull and process messages in a loop.
        // For real-time consumption, use multiple threads to pull concurrently.
        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 {
                    consumer.ack(message);
                    System.out.println("Message is acknowledged successfully, messageId= " + messageId);
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        }
        // Shut down when no longer needed
        // consumer.close();
    }
}