ApsaraMQ for RocketMQ:Scheduled and delayed messages

Send a scheduled message

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();
        }
    }
}

Consume with a push consumer

A push consumer receives messages through a message listener callback. The broker pushes messages to the consumer as they become ready.

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();
    }
}

Consume with a simple consumer

A simple consumer pulls messages from the broker and explicitly acknowledges each one after processing.

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();
    }
}