Sending and Receiving Ordered Messages

In DMS for RocketMQ, ordered messages are consumed in the exact order that they are produced.

Ordered messages are ordered globally or on the partition level.

Globally ordered messages: There is only one queue in a specific topic. All messages in the queue will be published and subscribed to in the first in, first out (FIFO) order.
Partition-level ordered message: Messages within a queue in a specific topic are published and subscribed to in the FIFO order. A producer specifies a message group for each message. Messages in the same group are allocated to the same queue.

The only difference between globally ordered messages and partition-level ordered messages is the number of queues. The code is the same.

Before sending and receiving ordered messages, collect RocketMQ connection information by referring to Collecting Connection Information.

Notes and Constraints

The gRPC protocol is only supported by RocketMQ v5.x but not v4.8.0.
To receive and send orderly messages, ensure the topic message type is Orderly before connecting a client to a RocketMQ instance of v5.x.
When the gRPC protocol is used to connect to a RocketMQ instance, whether a consumer consumes messages in sequence depends not on the consumption code, but on whether ordered consumption is enabled for the consumer group. The code for ordered consumption is the same as that for normal consumption.

Preparing the Environment

You can connect open-source Java clients to DMS for RocketMQ. The recommended Java client version is 5.0.5.

Using Maven

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-client-java</artifactId>
    <version>5.0.5</version>
</dependency>

Sending Ordered Messages

Refer to the following sample code or obtain more sample code from ProducerFifoMessageExample.java.

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.SessionCredentialsProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.message.Message;
import org.apache.rocketmq.client.apis.producer.Producer;
import org.apache.rocketmq.client.apis.producer.SendReceipt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.nio.charset.StandardCharsets;


public class ProducerFifoMessageExample {
    private static final Logger log = LoggerFactory.getLogger(ProducerFifoMessageExample.class);

    public static void main(String[] args) throws ClientException, IOException {
        final ClientServiceProvider provider = ClientServiceProvider.loadService();

        String topic = "yourNormalTopics";
        // Specify the gRPC address or gRPC public address
        String endpoints = "yourEndpoints";
        // Enter your username and key. Hard-coded or plaintext username and key are risky. You are advised to store them in ciphertext in a configuration file or an environment variable. The code below is required only if ACL was enabled during instance creation.
        String accessKey = System.getenv("ROCKETMQ_AK");
        String secretKey = System.getenv("ROCKETMQ_SK");
        SessionCredentialsProvider sessionCredentialsProvider =
                new StaticSessionCredentialsProvider(accessKey, secretKey);

        ClientConfiguration clientConfiguration = ClientConfiguration.newBuilder()
                .setEndpoints(endpoints)
                // .enableSsl(false)  // This line is mandatory to create an instance with SSL set to PLAINTEXT. This line is optional to create an instance with SSL set to PERMISSIVE.
                // .setCredentialProvider(sessionCredentialsProvider)  // Set the credential provider if ACL has been enabled.
                .build();

        final Producer producer = provider.newProducerBuilder()
                .setClientConfiguration(clientConfiguration)
                .setTopics(topic)
                .build();
        byte[] body = "This is a FIFO message for Apache RocketMQ".getBytes(StandardCharsets.UTF_8);
        String tag = "yourMessageTagA";
        final Message message = provider.newMessageBuilder()
                .setTopic(topic)
                .setTag(tag)
                .setKeys("yourMessageKey")
                // Specify a message group. Messages in the same group are allocated to the same queue.
                .setMessageGroup("yourMessageGroup0")
                .setBody(body)
                .build();
        try {
            final SendReceipt sendReceipt = producer.send(message);
            log.info("Send message successfully, messageId={}", sendReceipt.getMessageId());
        } catch (Throwable t) {
            log.error("Failed to send message", t);
        }

        // Close the producer when it is no longer needed.
        producer.close();
    }
}