RocketMQ Consumer Scaling and Load Balancing Strategies

In an interview scenario, the interviewer asks whether adding consumers helps when RocketMQ messages are backlogged.

Answer: If the number of consumers is less than the number of MessageQueues, adding consumers can increase consumption speed; otherwise it has no effect.

Consumer pull delay occurs when ProcessQueue thresholds are exceeded: message count > 1000, size > 100MB, or offset difference > 2000 for non‑ordered consumption.

For ordered consumption, lock failure causes a fixed 3‑second delay.

Common reasons for slow consumer processing include complex business logic, slow DB queries, cache latency, or slow external service calls.

Mitigation strategies: async calls with retries, caching default results, fallback, or storing messages locally and acknowledging later.

When increasing consumers, consider external service throughput and local DB/cache pressure.

MessageQueue allocation is performed periodically (default every 20 s). RocketMQ provides six allocation strategies:

Average allocation – sort consumers, compute average queues per consumer, distribute remainder.

// AllocateMessageQueueAveragely example
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()));
}

Round‑robin allocation – iterate consumers and assign queues cyclically.

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

Custom allocation – specify exact queues for a consumer.

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

Machine‑room allocation – consumers only consume queues from designated data‑center rooms.

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

Queues are first filtered by room name, then distributed using the average strategy.

Nearby‑room allocation – also assigns queues from empty rooms to existing consumers.

Consistent‑hash allocation – place consumers on a hash ring and bind each queue to the nearest consumer.

// AllocateMessageQueueConsistentHash example
Collection
cidNodes = new ArrayList<>();
for (String cid : cidAll) {
    cidNodes.add(new ClientNode(cid));
}
ConsistentHashRouter
router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt);
List
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 help balance load and avoid consumer bottlenecks in RocketMQ deployments.