tencent cloud

Sequential Message
Download PDF
Last updated: 2025-07-15 17:46:17
Sequential Message
Last updated: 2025-07-15 17:46:17
Download PDF
Sequential message is an advanced type of message offered by TDMQ for RocketMQ. For a specified Topic, messages are published and consumed in strict accordance with the principle of First-In-First-Out (FIFO). That is, messages sent first are consumed first, and messages sent later are consumed later.
Sequential messages are suitable for scenarios that have strict requirements on the sequence of message sending and consumption.

Use Cases

The comparison between sequential messages and general messages is as follows:
Message Type
Consumption Sequence
Performance
Use Cases
General message
No sequence
High
Suitable for scenarios with high demands for throughput and no requirement for production and consumption sequence.
Sequential message
Messages in a specified Topic following the FIFO rule
Average
Scenarios having average demands for throughput and requiring publishing and consuming all messages in a specified Topic in strict accordance with the FIFO rule
Sequential messages are often used in the following actual business scenarios:
Order creation: In some e-commerce systems, the creation, payment, refund, and logistics messages of an order must be produced or consumed in strict sequence. Otherwise, the order status delivery will be messed up, affecting normal businesses. Therefore, the messages of this order must be produced and consumed in a certain sequence in the client and message queue. In addition, the messages are sequentially dependent, and the processing of the next message depend on the processing result of the preceding message.
Log synchronization: In the scenario of sequential event processing or real-time incremental data synchronization, sequential messages can also play a vital role. For example, it is necessary to ensure that database operations are in sequence when MySQL binlogs are synchronized.
Financial scenarios: In some matchmaking transaction scenarios such as certain securities trading, priority is given to those who bid first case of the same biding price, so it is necessary to produce and consume sequential messages in a FIFO manner.

How It Works

In TDMQ for RocketMQ, the principle of sequential messages is shown in the figure below. You can partition messages according to a certain standard, such as ShardingKey in the figure. Messages of the same ShardingKey will be assigned to the same queue and consumed in sequence.



The code for sequential messages is as follows:
public class Producer {
    public static void main(String[] args) throws UnsupportedEncodingException {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
            producer.start();

            String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
            for (int i = 0; i < 100; i++) {
                int orderId = i % 10;
                Message msg =
                    new Message("TopicTest", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Integer id = (Integer) arg;
                        int index = id % mqs.size();
                        return mqs.get(index);
                    }
                }, orderId);

                System.out.printf("%s%n", sendResult);
            }

            producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
The main difference here is that the SendResult send(Message msg, MessageQueueSelector selector, Object arg) method is called. MessageQueueSelector is a queue selector and arg is a Java object, which can be passed in as the classification standard for message sending partition.
The MessageQueueSelector API is as follows:
public interface MessageQueueSelector {
    MessageQueue select(final List<MessageQueue> mqs, final Message msg, final Object arg);
}
Where, mqs is the queue that can be sent, msg is the message, arg is the object passed in the preceding send API, and the queue to which the message needs to be sent is returned. In the preceding sample, orderId is used as the partition classification standard, and the remainder of the number of all queues is obtained to send messages with the same orderId to the same queue.
In the production environment, it is recommended that you select the most fine-grained partition key for splitting. For example, when the order ID and user ID are used as the partition key keywords, the messages of the same end user will be processed in sequence, while those of different users will not.
Note:
To ensure high availability of messages, TDMQ for RocketMQ currently does not support "globally sequential messages" in a single queue (users who have already created globally sequential messages can continue to use them normally). If you want to ensure global sequence, you can use a consistent ShardingKey.

Consuming Sequential Messages

Sequential consumption code is as follows:
/**
 * Description: Sequential consumer
 */
public class OrderConsumer {
    /**
     *topic name
     */
    private static final String TOPIC_NAME = "order_topic";

    /**
     * consumer group name
     */
    private static final String GROUP_NAME = "group2";
    public static void main(String[] args) throws Exception {
        create message consumer
        // Instantiate the consumer.
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(GROUP_NAME,
                new AclClientRPCHook(new SessionCredentials("accessKey", "secretKey")),
                new AllocateMessageQueueAveragely(), true, null);
        // Set the NameServer address.
        consumer.setNamesrvAddr("rmq-xxx.rocketmq.xxxtencenttdmq.com:8080");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        /*
         Set whether the Consumer starts consuming from the head or tail of the queue on first startup
         if not first startup, then resume consumption from last position
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        // subscribe to all messages in the topic
        consumer.subscribe(TOPIC_NAME, "*");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                for (MessageExt msg : msgs) {
                    // You can see each queue has a unique consume thread to consume, order is ordered per queue (partition)
                    System.out.println("consumeThread=" + Thread.currentThread().getName() + ", queueId=" + msg.getQueueId() + ", msgId=" + msg.getMsgId() + ", content:" + new String(msg.getBody()));
                }
                try {
                    // Simulating business logic processing...
                    TimeUnit.SECONDS.sleep(1);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

Was this page helpful?
You can also Contact Sales or Submit a Ticket for help.
Yes
No

Feedback