How RocketMQ Implements Ordered Consumption Locking

Two‑layer protection for ordered consumption

The client guarantees that only one consumption task runs for a given MessageQueue. The first layer is the consuming flag inside ProcessQueue. After the pull thread stores messages via putMessage, it checks the flag; if consuming == false it flips the flag to true and signals the upper layer to submit a new ConsumeRequest to the thread pool. If the flag is already true, no new task is submitted, so newly pulled messages are merely placed into the internal TreeMap and wait.

Only one consumption task runs for a given queue at any moment.

The second layer is MessageQueueLock, which holds a distinct Object lock for each queue in a ConcurrentHashMap. When a consumption task starts it acquires the queue‑specific lock and enters a synchronized block. This lock acts as a safety net: even if two tasks somehow coexist, they serialize on the synchronized block, ensuring exclusive execution of the consumption logic.

while‑loop: lock holds the queue and consumes sequentially

Inside the synchronized block, ConsumeRequest.run() executes a while loop. Each iteration performs three steps:

Retrieve a batch of messages from ProcessQueue in offset order.

Invoke the registered MessageListenerOrderly to run business logic.

Based on the listener’s return value, decide whether to continue or stop. ProcessQueue stores messages in a TreeMap<Long, MessageExt> keyed by offset, guaranteeing natural ordering. By default consumeMessageBatchMaxSize = 1, so only one message is taken per iteration.

If the queue still contains messages, the variable continueConsume remains true and the loop proceeds. When the queue becomes empty, the consuming flag is reset to false, the loop exits, and the lock is released.

Consequently, logs such as

[ConsumeMessageThread_10] msg is :Hello RocketMQ 75
[ConsumeMessageThread_10] msg is :Hello RocketMQ 79
[ConsumeMessageThread_10] msg is :Hello RocketMQ 83
[ConsumeMessageThread_10] msg is :Hello RocketMQ 87
[ConsumeMessageThread_10] msg is :Hello RocketMQ 91
[ConsumeMessageThread_10] msg is :Hello RocketMQ 95
[ConsumeMessageThread_10] msg is :Hello RocketMQ 99

show a single thread processing multiple messages sequentially under the same lock, not concurrently.

Lock granularity is at the queue level, not per message. The lock is held for the entire loop, which differs from the imagined “lock per message, unlock, then lock next” model, but both achieve ordered processing.

Why autoCommit(false) does not stop the loop

The test code set context.setAutoCommit(false) and returned ConsumeOrderlyStatus.SUCCESS. The client still continued to the next message because autoCommit only controls whether the consumption offset is reported to the broker.

In ConsumeMessageOrderlyService.processConsumeResult:

If autoCommit == false and the status is SUCCESS, the method updates TPS statistics and leaves continueConsume as its default true. No ProcessQueue.commit() is invoked.

If autoCommit == true and the status is SUCCESS, the method additionally calls ProcessQueue.commit() to report the offset.

The effect of autoCommit is therefore limited to offset submission; it does not affect loop continuation.

Return‑status behavior (condensed from the original table):

SUCCESS : with autoCommit=true – commit offset and continue; with autoCommit=false – do not commit but continue.

COMMIT : commit offset and continue (independent of autoCommit).

ROLLBACK : with autoCommit=true – treated as SUCCESS with a warning; with autoCommit=false – message is returned to the queue and the current queue is paused.

Thus, a SUCCESS result always leads to the next message being consumed. Disabling autoCommit while returning SUCCESS causes the offset never to be committed; after a consumer restart the broker will redeliver those messages, resulting in duplicates. When autoCommit == false is required, the correct practice is to return ConsumeOrderlyStatus.COMMIT after successful processing.

Answer to the original expectation

The expectation of “take one message, lock the queue, consume, unlock, then take the next” is satisfied, but the implementation locks the entire queue and processes messages in a loop until the queue is empty. This design avoids the overhead of acquiring and releasing a lock for every single message, which would be costly at high throughput.

Summary

Ordered consumption in RocketMQ relies on two coordinated mechanisms:

The ProcessQueue.consuming flag prevents duplicate task submission for the same queue.

A synchronized queue‑specific lock ( MessageQueueLock) guarantees exclusive access to the consumption logic.

Once the lock is acquired, a while loop processes messages one by one in offset order, releasing the lock only when the queue becomes empty. The autoCommit flag merely decides whether the offset is reported to the broker; it does not control loop continuation. Misunderstanding this flag can lead to duplicate consumption if autoCommit is disabled while returning SUCCESS. Proper ordered processing also depends on the producer routing related messages to the same queue (e.g., using a consistent hash key).