How RocketMQ’s Evolving Batch Models Boost Throughput and Reduce Latency

Batch Processing in RocketMQ

Batch processing groups multiple messages into a single send operation, reducing per‑message overhead and improving throughput.

Early Producer‑Broker Batch Model

RocketMQ originally supported sending a collection of Message objects:

SendResult send(Message msg);
SendResult send(Collection<Message> msgs);

Constraints for the collection:

All messages must share the same Topic.

Cannot be a RetryTopic.

Cannot be scheduled (delayed) messages.

All messages must have the same isWaitStoreMsgOK flag.

The broker selects a queue based on the batch attributes, unpacks the batch, and processes each message as if sent individually, preserving full compatibility with single‑message semantics.

Index‑Build Pipeline Refactor

The original index‑building scans the CommitLog file serially, creates a ConsumeQueue entry per message, and performs costly checks, becoming a bottleneck at high throughput.

A lightweight DispatchRequestOrderlyQueue implements “unordered enqueue, ordered dequeue”, allowing batch‑level processing while keeping message order.

BatchCQ Model

BatchCQ is a special Topic type that stores multiple messages in a single slot, reducing the number of index entries. The topic attribute is set as follows:

public static final EnumAttribute QUEUE_TYPE_ATTRIBUTE =
    new EnumAttribute("queue.type", false,
        newHashSet("BatchCQ", "SimpleCQ"), "SimpleCQ");

topicConfig.getAttributes().put("+" + TopicAttributes.QUEUE_TYPE_ATTRIBUTE.getName(),
    "BatchCQ");

With BatchCQ, a batch of three messages and a batch of two messages occupy only two slots instead of five. Lookup uses binary search (O(log n)) rather than constant‑time slot calculation.

AutoBatch Model

AutoBatch automates classification and packaging while exposing the classic send(Message) API. Enable it with a single flag and optional tuning parameters:

// Enable AutoBatch
rmqProducer.setAutoBatch(true);
// Maximum batch size (bytes)
rmqProducer.batchMaxBytes(32 * 1024);
// Maximum wait time before sending a batch (ms)
rmqProducer.batchMaxDelayMs(10);
// Upper bound for total in‑memory batch size (bytes)
rmqProducer.totalBatchMaxBytes(32 * 1024 * 1024);

AutoBatch runs a background thread that flushes buffered messages when either the size or delay threshold is reached, keeping the serialization path lightweight.

Server‑Side Index‑Build Optimization

To sustain the higher production rate, the index‑build pipeline can be parallelized. Relevant broker configuration:

// Enable concurrent ConsumeQueue building
enableBuildConsumeQueueConcurrently=true
// In‑memory thresholds
maxTransferBytesOnMessageInMemory=256M
maxTransferCountOnMessageInMemory=32K
// Disk thresholds
maxTransferBytesOnMessageInDisk=64M
maxTransferCountOnMessageInDisk=32K

Performance Benchmarks

Benchmarks were run with OpenMessaging‑Benchmark on two Alibaba Cloud instance types (32‑core, 8‑core) using OpenJDK 11. Key results:

Early batch model without index optimization: ~80 k TPS.

Enabling AutoBatch raises throughput to ~270 k TPS (≈300 % increase).

BatchCQ topics achieve ~250 k TPS with P99 latency ≈264 ms, comparable to Kafka (~267 k TPS) but with significantly lower latency.

BatchCQ maintains performance with 10 000 partitions, whereas Kafka fails under the same configuration.

Summary

RocketMQ’s batch processing has evolved from a simple producer‑side collection to the BatchCQ storage model and the fully automated AutoBatch mechanism. Combined with the index‑build pipeline refactor, these changes deliver multi‑hundred‑kilo‑TPS throughput and low tail latency for cloud‑native messaging workloads.