🌐 中文 English
This article aims to explain the implementation principles of RocketMQ’s messages, mainly covering delayed messages, transactional messages, and ordered messages.
What is a Message
A message is a medium for communication between systems, used to transmit data or commands in distributed systems. Message transmission is achieved through message middleware, such as RocketMQ, Kafka, etc. Message middleware helps systems achieve high-performance, highly reliable data transmission through asynchronous decoupling.
RocketMQ primarily supports the following message models:
- Normal Messages: The most common message type, where producers send messages to queues, and consumers pull messages from queues for processing.
- Delayed Messages: Allow messages to be consumed by consumers only after a specified time.
- Transactional Messages: Support distributed transactions, achieving transactional consistency between message sending and business execution.
- Ordered Messages: Ensure messages are consumed by consumers in the same order they were sent by the producer. This article will focus on the implementation principles of delayed messages, transactional messages, and ordered messages.
Delayed Messages
Delayed messages are delivered immediately by the producer, rather than being sent to the Broker by the producer only when the time arrives. This shifts the pressure to the Broker, causing it to accumulate a large number of messages, which helps with peak shaving and valley filling. To prevent queue message congestion in the Broker or offset jumping around (preloading based on the principle of locality) and to avoid requiring too many ‘alarms’ to handle each message individually, which would degrade performance, RocketMQ directly defines 18 levels of delayed delivery: 1s, 5s, 10s, 30s, 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h. It then establishes a shared scheduled task thread pool with exactly 18 core threads to manage the delivery of all delayed messages. This way, delayed messages are uniformly categorized and constrained, making them easier to manage and allocate.
To achieve this, RocketMQ creates a dedicated Topic (SCHEDULE_TOPIC_XXXX) for storing delayed messages. Delayed messages are first sent to this Topic, which can reuse the commitLog and distribute messages to the consumeQueue. Consumers do not subscribe to this Topic, so they cannot consume these messages at this stage. The thread pool then schedules the messages in each queue under this Topic at regular intervals. Once a message expires, it is distributed to the original Topic for consumers to consume.
Each delay level corresponds to an independent task (DeliverDelayedMessageTimerTask) with a delay time of 100ms. These tasks are submitted to the thread pool. The task’s content involves obtaining the corresponding consumeQueue based on the incoming QueueID, along with the corresponding offset. The Broker periodically saves the consumption offset of the consumeQueue under SCHEDULE_TOPIC_XXXX. If the thread pool (deliverExecutorService) size is sufficient (e.g., equal to or greater than the number of delay levels), these tasks may be processed in parallel.
If a task expires, a new task is immediately created and thrown into the thread pool with a delay of 100ms. The task’s parameters include the updated offset, so the thread continues to consume subsequent messages, repeating this cycle. If the retrieved delayed message has not yet reached its scheduled time, the offset remains unchanged, and a new task is immediately created and added to the thread pool. This way, after 100ms, it will check again whether the message has expired.
From an implementation perspective, these methods greatly reduce the complexity of developing delayed messages. However, this implementation is inaccurate in terms of delay time. Task execution time and queue consumption wait times can cause the delay time to be inaccurate.
Transactional Messages
When talking about transactions, we refer to ACID (Atomicity, Consistency, Isolation, and Durability). For data operations within the same database, the database’s own mechanisms can ensure transactions. However, for data operations across different databases, the database’s own transactions cannot be used, and distributed transactions must be employed.
There are many solutions for implementing distributed transactions, such as 2PC, 3PC, TCC, local messages, and transactional messages. This article focuses on transactional messages.
Transactional messages are more suitable for asynchronous update scenarios, ensuring eventual consistency while allowing temporary inconsistency between the two ends of the data.
At the start of a transaction, a half message (half message) is first sent to the Broker. This is an incomplete message that consumers will skip. Then, based on the execution result of the local transaction, a decision is made to either send a commit message or a rollback message to the Broker. RocketMQ designs a check-back mechanism: the Broker queries the producer that sent the message through an exposed interface to check whether the transaction succeeded. In case of unexpected situations, rollback messages are used to ensure transaction consistency. If the sending producer crashes, other producers in the same producer group can be queried by the Broker, which is one of the roles of the producer group.
When sending a half message, it is not sent to the original Topic but to a specific Topic: RMQ_SYS_TRANS_HALF_TOPIC, with the default queue being 0. The Broker runs a scheduled thread service called TransactionalMessageCheckService, which periodically scans messages under the RMQ_SYS_TRANS_HALF_TOPIC Topic. It requests the producer’s check-back interface to see if the transaction succeeded. If successful, it restores the original Topic for consumers to consume. If failed, it does not redeliver the message.
Ordered Messages
Ordered messages require ensuring orderliness in three stages: sending, storage, and consumption, i.e., guaranteeing their temporal order or causal order.
In the sending stage, it must be ensured that a single producer and within a single producer, ordered messages are sent in a single-threaded (serial) manner to prevent order issues caused by multiple producers or multithreading.
In the storage stage, it must be ensured that related messages are stored sequentially within a single queue, ensuring that all related ordered messages are allocated to the same consumeQueue. This is achieved by taking the remainder of the orderId divided by the number of queues, so that the same order is always sent to the same queue, specifying the queue.
Consumption requirements are largely consistent with the sending stage. However, in terms of business logic, ordered message processing requires that if a preceding message fails to be consumed, all related subsequent messages also fail directly. These messages can then be persistently stored elsewhere and re-consumed after fixes are applied. Additionally, there is consumer load balancing.
For consumption scenarios, RocketMQ uses three common types of locks to ensure message consumption order as much as possible.
- Distributed Lock: Binds the corresponding queue to a consumer. If it is found to be already bound to another consumer, messages cannot be pulled for consumption. The consumer periodically (default every 20s) renews this lock to maintain ownership of the distributed lock.
- Synchronized: Ensures that only one thread consumes from this queue at any given time (allowing the same consumer to concurrently consume messages from different queues).
- ReentrantLock: This lock acts more like a flag, indicating that the current queue still has messages being consumed, preventing rebalancing and waiting for the next rebalancing cycle. (This lock offers finer granularity.)
The dialectical relationship between orderliness and availability: If absolute orderliness must be guaranteed, availability must be sacrificed, meaning messages cannot be sent to different queues. Conversely, if availability must be guaranteed, absolute orderliness cannot be ensured.
RocketMQ provides solutions for both modes: For absolute orderliness, when creating a Topic, the -o parameter (--order) must be set to true, and the configurations orderMessageEnable and returnOrderTopicConfigToBroker in NameServer must also be true. If any of these conditions is false, availability is chosen.
Summary
The above is my explanation of the implementation principles of RocketMQ’s three special message types.
