ordered message architecture - ApsaraMQ for RocketMQ - Alibaba Cloud Documentation Center

Ordered messages are an advanced feature in ApsaraMQ for RocketMQ. This topic describes the scenarios, working principle, limits, usage, and best practices for ordered messages.

Scenarios

In scenarios such as ordered event processing, transaction matchmaking, and real-time incremental data synchronization, heterogeneous systems must maintain strong consistency through state synchronization. In these scenarios, event changes in upstream systems must be transmitted to downstream systems for sequential processing. You can use ordered messages in ApsaraMQ for RocketMQ to ensure sequential data transmission.

Scenario 1: Transaction matchmaking

交易撮合

In a securities and stock transaction matchmaking scenario, if multiple bids have the same price, the first-in, first-out (FIFO) principle applies. The downstream order processing system must process orders strictly in the order that the bids were placed.

Scenario 2: Real-time incremental data synchronization

In a database change incremental synchronization scenario, the upstream source database performs add, delete, and modify operations. The binary operation logs are sent as messages through ApsaraMQ for RocketMQ to a downstream search system. The downstream system then sequentially restores the data from the messages to refresh its state. If you use normal messages, the state may become inconsistent with the expected operation results. Ordered messages ensure that the downstream state is consistent with the upstream operation results.

How it works

What are ordered messages?

Ordered messages are an advanced message type in ApsaraMQ for RocketMQ. They allow consumers to receive messages in the same order in which they were sent. This enables sequential processing in your business scenarios.

Unlike other message types, ordered messages guarantee the sequential relationship between multiple messages during sending, storage, and delivery.

In ApsaraMQ for RocketMQ, the order of messages is defined by message groups. When you send ordered messages, you must specify a message group for each message.

Important

Order is guaranteed only for messages within the same message group. The order of messages in different message groups or messages without a specified group is not guaranteed.

The message group-based ordering logic allows for fine-grained splitting based on your business logic. This can improve system parallelism and throughput while ensuring local business order.

How message order is ensured

Message ordering in ApsaraMQ for RocketMQ involves two stages: production order and consumption order.

Production order: ApsaraMQ for RocketMQ uses a protocol between the producer and the server to ensure that messages sent serially by a single producer are stored and persisted in that order.
To ensure the production order of messages, you must meet the following conditions:
- Same message group
  Production order applies only to a single message group. When a producer sends messages, it can set a message group for each message. Order is guaranteed only for messages within the same message group. The order of messages in different message groups or messages without a specified group is not guaranteed.
- Single producer
  Production order is guaranteed only for a single producer. If different producers send messages to the same message group, even if they are distributed across different systems, the message order cannot be guaranteed.
- Serial sending
  The ApsaraMQ for RocketMQ producer client supports multi-threaded secure access. However, if a producer uses multiple threads to send messages concurrently, the order of messages sent by different threads cannot be guaranteed.
When a producer that meets these conditions sends ordered messages to ApsaraMQ for RocketMQ, messages with the same message group are stored in the same queue in the order they were sent. The server's sequential storage logic is as follows:
- Messages from the same message group are stored in the same queue in sequential order.
- Messages from different message groups can be mixed in the same queue, but their continuity is not guaranteed.
As shown in the preceding figure, messages from Message Group 1 and Message Group 4 are mixed in Queue 1. ApsaraMQ for RocketMQ ensures that messages G1-M1, G1-M2, and G1-M3 from Message Group 1 are stored in the order they were sent. Messages G4-M1 and G4-M2 from Message Group 4 are also stored in order. However, the order between messages from Message Group 1 and Message Group 4 is not guaranteed.
Consumption order: ApsaraMQ for RocketMQ uses a protocol between the consumer and the server to ensure that messages are consumed strictly in the order in which they are stored.
To ensure the consumption order of messages, you must meet the following conditions:
- Delivery order
  ApsaraMQ for RocketMQ uses a client SDK and a server communication protocol to ensure that messages are delivered in the order they are stored. However, when your application consumes messages, it must strictly follow the receive-process-acknowledge semantics to prevent message disorder caused by asynchronous processing.
  Important
  If the consumer type is PushConsumer, ApsaraMQ for RocketMQ ensures that messages are delivered to the consumer one by one in the order they are stored. If the consumer type is SimpleConsumer, the consumer may pull multiple messages at once. In this case, your application must ensure the consumption order. For more information about consumer types, see Consumer types.
- Limited retries
  The delivery of ordered messages in ApsaraMQ for RocketMQ is limited to a specific number of retries. If a message consistently fails to be consumed and exceeds the maximum number of retries, it is no longer retried. The system skips this message to prevent it from blocking subsequent message processing.
  For scenarios that require strict consumption order, you can set a reasonable number of retries to prevent message disorder caused by incorrect parameter settings.

Combinations of production and consumption order

If messages must be processed strictly in FIFO order, both production and consumption order must be guaranteed. However, in typical business scenarios, a single producer may connect to multiple downstream consumers, not all of whom require ordered consumption. You can use different combinations of production and consumption order for various business scenarios. For example, you can send ordered messages but use concurrent consumption to increase throughput. The following table describes several combinations:

Production order	Consumption order	Ordering effect
Ordered sending (message group is set).	Ordered consumption	Strict message order is guaranteed at the message group level. The consumption order is identical to the sending order for messages within the same message group.
Ordered sending (message group is set).	Concurrent consumption	Concurrent consumption. Messages are processed as chronologically as possible, but order is not guaranteed.
Non-ordered sending (no message group is set).	Ordered consumption	Strict order at the queue level. Based on the queue properties of ApsaraMQ for RocketMQ, the consumption order matches the storage order within the queue, but it is not guaranteed to match the sending order.
Non-ordered sending (no message group is set).	Concurrent consumption	Concurrent consumption. Messages are processed as chronologically as possible, but order is not guaranteed.

Lifecycle of an ordered message

生命周期

Initialization
The message is built and initialized by the producer and is ready to be sent to the broker.
Pending consumption
The message is sent to the broker and is visible and available to the consumer.
Being consumed
The message is obtained by the consumer and processed based on the local business logic of the consumer.
In this process, the broker waits for the consumer to return the consumption result. If no response is received from the consumer in a specific period of time, ApsaraMQ for RocketMQ performs retries on the message. For more information, see Consumption retry.
Consumption commitment
The consumer completes the consumption and commits the consumption result to the broker. The broker marks whether the current message is consumed.
By default, ApsaraMQ for RocketMQ retains all messages. When the consumption result is committed, the message is marked as consumed instead of being immediately deleted. Messages are deleted only if the retention period expires or the system has insufficient storage space. Before a message is deleted, the consumer can re-consume the message.
Message deletion
If the retention period of message expires or the storage space is insufficient, ApsaraMQ for RocketMQ deletes the earliest stored messages from the physical file in a rolling manner. For more information, see Message storage and cleanup.

Important

When a message fails to be consumed or consumption times out, the server's retry logic is triggered. The retried message is considered a new message, and the lifecycle of the original message ends.
To ensure message order, when a failed ordered message is retried, subsequent messages in the same message group are blocked. They can be consumed only after the current message is successfully consumed.

Limits

You can send ordered messages only to topics whose MessageType is set to FIFO. The message type must match the topic type.

Consumption concurrency optimization

Your client must use the RocketMQ 5.x gRPC SDK. When consuming ordered messages, the PushConsumer client in this SDK series can assign messages from the same MessageQueue to different threads for concurrent consumption based on their MessageGroup. This feature significantly improves the concurrency of ordered message consumption. The improvement is more significant when MessageGroup values are more discrete. The supported SDK versions are as follows:

Java SDK: 5.0.8 and later.
C++ SDK: 5.0.3 and later.
Other SDKs: Not supported.

Examples

Unlike sending normal messages, you must set a message group when you send an ordered message. You can design message groups with the finest possible granularity based on your business scenario. This facilitates business logic splitting and improves concurrent scaling.

The following sample code shows how to send and receive ordered messages in Java:

For the complete sample code for sending and receiving messages, see RocketMQ 5.x gRPC SDK.

Sample code

Send ordered messages

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 {
        /**
         * The endpoint of the instance. You can obtain the endpoint from the Endpoint tab on the Instance Details page in the console.
         * If you access the instance from an Alibaba Cloud ECS instance over the internal network, we recommend that you use the VPC endpoint.
         * If you access the instance from your local computer or a data center over the internet, you can use the public endpoint. To use the public endpoint, you must enable public network access for the instance.
         */
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // The name of the destination topic for the message. You must create the topic in the console in advance. An error is returned if you use a topic that does not exist.
        String topic = "Your Topic";
        ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * If you use the public endpoint, you must also configure the username and password for the instance. You can obtain the username and password from the Intelligent Authentication tab of the Access Control page for the instance in the console.
         * If you access the instance from an Alibaba Cloud ECS instance over the internal network, you do not need to configure this. The server automatically obtains the information based on the VPC.
         * If the instance is a serverless instance, you must set the username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
         */
        // builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));
        ClientConfiguration configuration = builder.build();
        Producer producer = provider.newProducerBuilder()
                .setTopics(topic)
                .setClientConfiguration(configuration)
                .build();
         // Send an ordered message.
        Message message = provider.newMessageBuilder()
                .setTopic("topic")
                // Set the message key. You can use the key to accurately find a specific message.
                .setKeys("messageKey")
                // Set the message tag. The consumer can use the tag to filter messages.
                .setTag("messageTag")
                // Set the message group for the ordered message. Keep this group as discrete as possible to avoid hot spot message groups.
                .setMessageGroup("fifoGroup001")
                // Message body.
                .setBody("messageBody".getBytes())
                .build();
        try {
            // Send the message. Pay attention to the sending result and catch exceptions such as failures.
            SendReceipt sendReceipt = producer.send(message);
            System.out.println(sendReceipt.getMessageId());
        } catch (ClientException e) {
            e.printStackTrace();
        }
    }
}

Consume messages with PushConsumer

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 {
        /**
         * The endpoint of the instance. You can obtain the endpoint from the Endpoint tab on the Instance Details page in the console.
         * If you access the instance from an Alibaba Cloud ECS instance over the internal network, we recommend that you use the VPC endpoint.
         * If you access the instance from your local computer or a data center over the internet, you can use the public endpoint. To use the public endpoint, you must enable public network access for the instance.
         */
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // The destination topic that you want to subscribe to. You must create the topic in the console in advance. An error is returned if you use a topic that does not exist.
        String topic = "Your Topic";
        // The consumer group to which the consumer belongs. You must create the group in the console in advance. An error is returned if you use a group that does not exist.
        String consumerGroup = "Your ConsumerGroup";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * If you use the public endpoint, you must also configure the username and password for the instance. You can obtain the username and password from the Intelligent Authentication tab of the Access Control page for the instance in the console.
         * If you access the instance from an Alibaba Cloud ECS instance over the internal network, you do not need to configure this. The server automatically obtains the information based on the VPC.
         * If the instance is a serverless instance, you must set the username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
        ClientConfiguration clientConfiguration = builder.build();
        // The filter expression for message subscription. This indicates that all messages with any tag are subscribed.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Initialize PushConsumer. You must attach the consumer group, communication parameters, and subscription relationship.
        // When you consume ordered messages, ensure that the current consumer group is in ordered delivery mode. Otherwise, messages are still delivered concurrently and out of order.
        PushConsumer pushConsumer = provider.newPushConsumerBuilder()
                .setClientConfiguration(clientConfiguration)
                // Set the consumer group.
                .setConsumerGroup(consumerGroup)
                // Set the pre-attached subscription relationship.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression))
                // Set the message listener.
                .setMessageListener(messageView -> {
                    // Process the message and return the consumption result.
                    // LOGGER.info("Consume message={}", messageView);
                    System.out.println("Consume Message: " + messageView);
                    return ConsumeResult.SUCCESS;
                })
                .build();
        Thread.sleep(Long.MAX_VALUE);
        // If you no longer need the PushConsumer, you can shut down the process.
        //pushConsumer.close();
    }
}

Consume messages with SimpleConsumer

import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientConfigurationBuilder;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
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.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 {
        /**
         * The endpoint of the instance. You can obtain the endpoint from the Endpoint tab on the Instance Details page in the console.
         * If you access the instance from an Alibaba Cloud ECS instance over the internal network, we recommend that you use the VPC endpoint.
         * If you access the instance from your local computer or a data center over the internet, you can use the public endpoint. To use the public endpoint, you must enable public network access for the instance.
         */
        String endpoints = "rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080";
        // The destination topic that you want to subscribe to. You must create the topic in the console in advance. An error is returned if you use a topic that does not exist.
        String topic = "Your Topic";
        // The consumer group to which the consumer belongs. You must create the group in the console in advance. An error is returned if you use a group that does not exist.
        String consumerGroup = "Your ConsumerGroup";
        final ClientServiceProvider provider = ClientServiceProvider.loadService();
        ClientConfigurationBuilder builder = ClientConfiguration.newBuilder().setEndpoints(endpoints);
        /**
         * If you use the public endpoint, you must also configure the username and password for the instance. You can obtain the username and password from the Intelligent Authentication tab of the Access Control page for the instance in the console.
         * If you access the instance from an Alibaba Cloud ECS instance over the internal network, you do not need to configure this. The server automatically obtains the information based on the VPC.
         * If the instance is a serverless instance, you must set the username and password for public network access. If authentication-free access over the internal network is enabled, you do not need to set the username and password for internal network access.
         */
        //builder.setCredentialProvider(new StaticSessionCredentialsProvider("Instance UserName", "Instance Password"));        
        ClientConfiguration clientConfiguration = builder.build();

        Duration awaitDuration = Duration.ofSeconds(10);
        // The filter expression for message subscription. This indicates that all messages with any tag are subscribed.
        String tag = "*";
        FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
        // Initialize SimpleConsumer. You must attach the consumer group, communication parameters, and subscription relationship.
        // When you consume ordered messages, ensure that the current consumer group is in ordered delivery mode. Otherwise, messages are still delivered concurrently and out of order.
        SimpleConsumer consumer = provider.newSimpleConsumerBuilder().setClientConfiguration(clientConfiguration)
                // Set the consumer group.
                .setConsumerGroup(consumerGroup)
                // Set the long polling timeout period.
                .setAwaitDuration(awaitDuration)
                // Set the pre-attached subscription relationship.
                .setSubscriptionExpressions(Collections.singletonMap(topic, filterExpression)).build();
        // Set the maximum number of messages to pull this time.
        int maxMessageNum = 16;
        // Set the invisible duration for the messages.
        Duration invisibleDuration = Duration.ofSeconds(10);
        // SimpleConsumer requires the client to actively loop to get messages and process them.
        // To improve real-time consumption, we recommend using multiple threads to pull messages concurrently.
        while (true) {
            // Note that for messages in the same MessageGroup, if a preceding message is not consumed, you cannot get subsequent messages by calling Receive again.
            final List<MessageView> messageViewList = consumer.receive(maxMessageNum, invisibleDuration);
            messageViewList.forEach(messageView -> {
                System.out.println(messageView);
                // After consumption is complete, you must actively call ACK to submit the consumption result.
                try {
                    consumer.ack(messageView);
                } catch (ClientException e) {
                    // If the pull fails due to system throttling or other reasons, you need to re-initiate the request to get the message.
                    e.printStackTrace();
                }
            });
        }
        // If you no longer need the SimpleConsumer, you can shut down the process.
        // consumer.close();
    }
}

Obtain consumption retry logs for ordered messages

If ordered messages are consumed in Push mode, the messages are retried on the consumer client and the broker cannot obtain the details of the retry logs. If the delivery result displayed in the trace of an ordered message indicates that the message delivery failed, you must check the information about the maximum number of retries and the consumer client in the consumer client logs.

For information about the log path of a consumer client, see Log configuration.

You can search the following keywords to query the information about consumption failures in client logs:

Message listener raised an exception while consuming messages
Failed to consume fifo message finally, run out of attempt times

Best practices

Use serial consumption to prevent out-of-order processing

We recommend that you consume messages serially instead of in batches. Batch consumption can cause messages to be processed out of order.

For example, messages are sent in the order 1, 2, 3, and 4. During batch consumption, the consumption sequence might be 1, then a batch of [2, 3] that fails, then a retry of [2, 3], and finally 4. In this case, the failure of message 3 might cause message 2 to be processed repeatedly, leading to out-of-order consumption.

Distribute message groups to avoid hot spots

ApsaraMQ for RocketMQ ensures that messages in the same message group are stored in the same queue. If messages from different business scenarios are concentrated in a few message groups, or even a single message group, the storage pressure is focused on a few queues, or a single queue, on the server. This can create performance hot spots and hinder scalability. A common design practice is to use order IDs or user IDs as the basis for ordering. This ensures that messages for the same end user are processed in order, while messages for different users do not need to be processed in order.

Therefore, we recommend that you define message groups at a fine granularity. For example, you can use order IDs or user IDs as message group keys. This ensures that messages related to the same end user are processed in order, while allowing messages for different users to be processed in parallel.