Understanding the Full Lifecycle of a RocketMQ Message: From Production to Deletion

Core Concepts

NameServer : a registration center that stores topic routing information and manages brokers; NameServers do not communicate with each other.

Broker : the core component that stores messages; multiple broker instances with the same name form a broker group.

Topic : a logical collection of messages, which can be distributed across different broker groups.

Queue : each topic can have many queues (default 4 per broker group).

Producer : the entity that creates and sends messages.

Producer Group : a logical grouping of producers.

Consumer : the entity that receives messages.

Consumer Group : a logical grouping of consumers; different groups consume independently.

Message Birth and Sending

Producers create a DefaultMQProducer , set the NameServer address, start the producer, and send a Message :

<code>DefaultMQProducer producer = new DefaultMQProducer("sanyouProducer");
producer.setNamesrvAddr("localhost:9876");
producer.start();
Message msg = new Message("sanyouTopic", "TagA", "三友的java日记 ".getBytes(RemotingHelper.DEFAULT_CHARSET));
SendResult sendResult = producer.send(msg);
</code>

Key questions include how the producer discovers broker addresses, how a queue is chosen, and what happens on send failure.

Routing Table

When a broker starts, it registers its IP, port, topics, and queue information with the NameServer, forming a routing table.

Producers pull the routing table from the NameServer at startup and refresh it every 30 seconds.

Queue Selection

RocketMQ provides two built‑in queue selection algorithms:

Round‑robin (default): distributes messages evenly across queues.

Latency‑fault‑tolerant: prefers queues with the smallest delivery latency.

Enable the latency‑fault algorithm:

<code>producer.setSendLatencyFaultEnable(true);
</code>

Custom selectors can be implemented by providing a MessageQueueSelector implementation.

<code>SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
    @Override
    public MessageQueue select(List&lt;MessageQueue&gt; mqs, Message msg, Object arg) {
        // select a queue
        return null;
    }
}, new Object());
</code>

RocketMQ also offers three built‑in selectors: random, hash, and rack‑aware (empty implementation).

Send Exception Handling

If a broker crashes or a send times out, RocketMQ retries (default two times). The retry count can be increased:

<code>producer.setRetryTimesWhenSendFailed(10);
</code>

Large Message Handling

Messages larger than the default 4 KB are compressed before being sent to reduce network usage.

Message Storage

CommitLog

All messages are persisted in a CommitLog file; each broker has a single CommitLog that is split into 1 GB segments.

When a segment fills up, a new file is created.

Zero‑Copy Techniques

To achieve high‑performance I/O, RocketMQ uses zero‑copy via mmap() and sendfile() .

mmap()

Memory‑maps a file into the process address space, allowing direct kernel‑buffer access.

<code>FileChannel fileChannel = new RandomAccessFile("test.txt", "rw").getChannel();
MappedByteBuffer mappedByteBuffer = fileChannel.map(FileChannel.MapMode.READ_WRITE, 0, fileChannel.size());
</code>

sendfile()

Transfers file data directly from kernel space to the socket, reducing both copies and context switches.

<code>FileChannel channel = FileChannel.open(Paths.get("./test.txt"), StandardOpenOption.WRITE, StandardOpenOption.CREATE);
channel.transferTo(position, len, target);
</code>

RocketMQ primarily uses the mmap‑based zero‑copy for its internal file I/O.

Flush Mechanisms

Asynchronous flush writes to the page cache and returns immediately; a background thread flushes to disk (default mode).

Synchronous flush waits for the data to be persisted before acknowledging the producer, offering higher reliability at the cost of throughput.

High Availability

Broker groups provide redundancy. Two modes are supported:

Master‑slave synchronous mode: the master (BrokerId 0) replicates topic metadata and messages to slaves.

Dledger mode: uses Raft to elect a leader; a majority of nodes must acknowledge a write for it to be considered successful.

Message Consumption

Consumers pull routing information from the NameServer, learn which queues belong to a subscribed topic, and then request messages from those queues.

<code>DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("sanyouConsumer");
</code>

Consumption Modes

Cluster mode : each message is consumed by only one consumer within a group (default).

Broadcast mode : every consumer in the group receives each message.

<code>consumer.setMessageModel(MessageModel.BROADCASTING);
</code>

ConsumeQueue

For each queue, RocketMQ maintains a ConsumeQueue that stores the offset and length of messages in the CommitLog, enabling fast lookup without scanning the entire log.

Message Ordering

To guarantee order, producers must send ordered messages to the same queue, the broker stores them sequentially, and consumers use MessageListenerOrderly to consume in order.

<code>consumer.registerMessageListener(new MessageListenerOrderly() {
    @Override
    public ConsumeOrderlyStatus consumeMessage(List&lt;MessageExt&gt; msgs, ConsumeOrderlyContext context) {
        // process ordered messages
        return null;
    }
});
</code>

Message Cleanup

CommitLog files are deleted when any of the following conditions are met:

Manual deletion.

Daily automatic cleanup at 04:00.

Disk usage reaches 75% (expire‑time based).

Disk usage reaches 85% (force delete regardless of age).

Files older than the configured expiration (default 72 hours) are considered expired.

Lifecycle Summary

Message Sending

Producer creates a message.

Producer fetches routing info.

Producer selects a queue via an algorithm.

Producer sends the message to the broker.

Message Storage

Broker writes the message to CommitLog.

Broker records the message’s position in a ConsumeQueue.

High Availability

In cluster mode, messages are replicated to slave brokers or Dledger followers.

Message Consumption

Consumer pulls routing info and selects queues based on its consumption mode.

Consumer connects to the broker and requests messages.

Broker uses the ConsumeQueue to locate messages in the CommitLog and returns them.

Message Cleanup

CommitLog files are removed according to the cleanup policies described above.

Final Note

https://github.com/sanyou3/rocketmq.git