When Does Adding Consumers Reduce RocketMQ Backlog?

In an interview setting the interviewer asks whether adding more consumers helps when a RocketMQ topic is backlogged. The answer depends on the relationship between the number of consumers and the number of MessageQueues for the topic.

If the consumer count is less than the MessageQueue count, adding a consumer can increase the pull frequency and reduce backlog. For example, a topic with 4 MessageQueues and 2 consumers will consume faster after a third consumer is added, as shown in the diagram.

When the consumer count is equal to or exceeds the MessageQueue count, adding more consumers brings no benefit because each MessageQueue is already assigned to a consumer.

Even in the first case, adding consumers does not always guarantee faster consumption. Pull delays can occur when the local ProcessQueue reaches certain thresholds: message count > 1000 (default), total size > 100 MB (default), or offset difference > 2000 for non‑ordered consumption. In ordered consumption, a lock‑failure also causes a 3‑second delayed pull.

The root cause of a delayed pull is often slow consumer processing. Typical reasons include complex business logic, slow database queries, sluggish cache (e.g., Redis) responses, or slow external service calls.

For slow external calls, the interview outlines three mitigation patterns: (1) use asynchronous calls with retries when the external call is merely a notification; (2) cache a default response and fall back to it on failure when the result must be processed; (3) store the message locally, return CONSUME_SUCCESS to the broker, and process the message after the external service recovers.

When scaling consumers, one must also consider the impact on the downstream system: a sudden increase in request volume may exceed the external service’s throughput limits, causing it to slow down or fail. Similarly, database or cache pressure can negate the benefits of added consumers.

RocketMQ assigns MessageQueues to consumers via a load‑balancing timer that runs every 20 seconds by default. Six allocation strategies are provided:

1. Average Allocation Strategy

Consumers are sorted, the average number of queues per consumer is calculated, and any remainder (mod) is distributed one by one to the first few consumers.

int index = cidAll.indexOf(currentCID);
int mod = mqAll.size() % cidAll.size();
int averageSize = mqAll.size() <= cidAll.size() ? 1 : (mod > 0 && index < mod ? mqAll.size() / cidAll.size() + 1 : mqAll.size() / cidAll.size());
int startIndex = (mod > 0 && index < mod) ? index * averageSize : index * averageSize + mod;
int range = Math.min(averageSize, mqAll.size() - startIndex);
for (int i = 0; i < range; i++) {
    result.add(mqAll.get((startIndex + i) % mqAll.size()));
}

2. Round‑Robin (Circle) Allocation Strategy

Iterate over consumers and assign each MessageQueue in turn; if queues outnumber consumers, multiple passes are made.

int index = cidAll.indexOf(currentCID);
for (int i = index; i < mqAll.size(); i++) {
    if (i % cidAll.size() == index) {
        result.add(mqAll.get(i));
    }
}

3. Custom Allocation Strategy

Developers can specify which queues a consumer should handle at startup.

AllocateMessageQueueByConfig allocateMessageQueueByConfig = new AllocateMessageQueueByConfig();
allocateMessageQueueByConfig.setMessageQueueList(Arrays.asList(new MessageQueue("messageQueue1", "broker1", 0)));
consumer.setAllocateMessageQueueStrategy(allocateMessageQueueByConfig);
consumer.start();

4. Machine‑Room Allocation Strategy

Consumers are bound to specific data‑center rooms; only queues from those rooms are considered.

AllocateMessageQueueByMachineRoom allocateMessageQueueByMachineRoom = new AllocateMessageQueueByMachineRoom();
allocateMessageQueueByMachineRoom.setConsumeridcs(new HashSet<>(Arrays.asList("room1", "room2")));
consumer.setAllocateMessageQueueStrategy(allocateMessageQueueByMachineRoom);
consumer.start();

5. Machine‑Room Nearby Allocation Strategy

After filtering queues by room, the remaining queues are distributed using the average allocation method, allowing rooms without dedicated consumers to receive queues.

6. Consistent‑Hash Allocation Strategy

Consumers are placed on a hash ring; each MessageQueue is hashed and bound to the nearest consumer node clockwise.

Collection<ClientNode> cidNodes = new ArrayList<>();
for (String cid : cidAll) {
    cidNodes.add(new ClientNode(cid));
}
ConsistentHashRouter<ClientNode> router = customHashFunction != null ?
    new ConsistentHashRouter<>(cidNodes, virtualNodeCnt, customHashFunction) :
    new ConsistentHashRouter<>(cidNodes, virtualNodeCnt);
List<MessageQueue> results = new ArrayList<>();
for (MessageQueue mq : mqAll) {
    ClientNode clientNode = router.routeNode(mq.toString());
    if (clientNode != null && currentCID.equals(clientNode.getKey())) {
        results.add(mq);
    }
}

These strategies illustrate how RocketMQ balances load across consumers, and the interview demonstrates the analytical reasoning behind choosing the appropriate approach.