Understanding RocketMQ Producer and Consumer Load Balancing Mechanisms

RocketMQ is a high‑performance distributed messaging middleware that uses a long‑polling pull model and can handle millions of queued messages on a single node. Its core components include Producer, Consumer, Broker, NameServer, ProducerGroup and ConsumerGroup, each playing a specific role in message production, storage, and consumption.

The system organizes messages by Topic , which are physically stored in MessageQueue instances on individual brokers. Because a topic may span multiple brokers and queues, load‑balancing is required both when producing and consuming messages.

Producer workflow :

Fetch routing information from the NameServer.

Parse the routing data into local structures ( TopicPublishInfo , MessageQueue list).

Select a target broker and queue based on the routing info and send the message.

public boolean updateTopicRouteInfoFromNameServer(final String topic) {
    // query NameServer, compare old and new RouteData, build TopicPublishInfo
    // ... (code omitted for brevity)
    return true;
}

The routing synchronization step ensures the producer knows which broker and queue to target. It involves querying the NameServer, detecting changes, and updating local tables.

public class TopicRouteData extends RemotingSerializable {
    private List
queueDatas;
    private List
brokerDatas;
    // getters/setters
}

After routing is resolved, the producer selects a MessageQueue using a round‑robin index, skipping the last failed broker:

public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
    for (int i = 0; i < messageQueueList.size(); i++) {
        int pos = Math.abs(sendWhichQueue.getAndIncrement()) % messageQueueList.size();
        MessageQueue mq = messageQueueList.get(pos);
        if (!mq.getBrokerName().equals(lastBrokerName)) return mq;
    }
    return selectOneMessageQueue();
}

Consumer workflow :

Periodically synchronize topic routing from the NameServer.

Build local subscription info ( TopicSubscribeInfo ).

Execute a rebalance to assign specific MessageQueue instances to each consumer in the group.

private void startScheduledTask() {
    scheduledExecutorService.scheduleAtFixedRate(() -> {
        try { updateTopicRouteInfoFromNameServer(); } catch (Exception e) { log.error(...); }
    }, 10, clientConfig.getPollNameServerInterval(), TimeUnit.MILLISECONDS);
}

The rebalance service runs in a background thread, invoking doRebalance() to recompute assignments:

public class RebalanceService extends ServiceThread {
    public void run() {
        while (!isStopped()) {
            waitForRunning(waitInterval);
            mqClientFactory.doRebalance();
        }
    }
}

During rebalance, the framework gathers all queues for a topic and all consumer IDs, sorts them, and applies an AllocateMessageQueueStrategy . The default average‑allocation algorithm distributes queues as evenly as possible:

public List
allocate(String consumerGroup, String currentCID, List
mqAll, List
cidAll) {
    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);
    List
result = new ArrayList<>();
    for (int i = 0; i < range; i++) {
        result.add(mqAll.get((startIndex + i) % mqAll.size()));
    }
    return result;
}

A concrete example with three brokers (a, b, c) each providing three queues for topic_demo shows how nine queues are divided among four consumers, yielding allocation results such as [broker_a_0, broker_a_1, broker_a_2] for the first consumer.

To force consumption on a specific machine, the allocation strategy can be customized to return an empty list for all consumers except the target IP. The modified AllocateMessageQueueAveragely checks the consumer’s CID (which contains the IP) before performing the normal calculation.

public List
allocate(String consumerGroup, String currentCID, List
mqAll, List
cidAll) {
    if (!cidAll.contains(currentCID)) return new ArrayList<>(); // only target IP proceeds
    // original average‑allocation logic follows
    ...
}

In summary, the article walks through RocketMQ’s architecture, the producer and consumer routing synchronization, queue selection, and rebalance processes, and demonstrates how to adapt the allocation algorithm to achieve machine‑specific consumption, providing both conceptual explanations and concrete Java code snippets.