墨与星辰

主题

负载均衡机制 ​

深入理解RocketMQ的负载均衡机制，掌握队列分配、负载均衡算法等核心功能，理解动态负载均衡、集群负载均衡等机制。

📋 负载均衡概览 ​

RocketMQ通过负载均衡机制实现消息队列的合理分配，保证消费者负载均衡。

🎯 一、队列分配 ​

1.1 AllocateMessageQueueStrategy ​

AllocateMessageQueueStrategy接口：

java
public interface AllocateMessageQueueStrategy {
    List<MessageQueue> allocate(String consumerGroup, String currentCID,
            List<MessageQueue> mqAll, List<String> cidAll);
    String getName();
}

1.2 平均分配 ​

AllocateMessageQueueAveragely：

java
@Override
public List<MessageQueue> allocate(String consumerGroup, String currentCID,
        List<MessageQueue> mqAll, List<String> cidAll) {
    if (currentCID == null || currentCID.length() < 1) {
        throw new IllegalArgumentException("currentCID is empty");
    }
    if (mqAll == null || mqAll.isEmpty()) {
        throw new IllegalArgumentException("mqAll is null or mqAll empty");
    }
    if (cidAll == null || cidAll.isEmpty()) {
        throw new IllegalArgumentException("cidAll is null or cidAll empty");
    }
    
    List<MessageQueue> result = new ArrayList<MessageQueue>();
    if (!cidAll.contains(currentCID)) {
        log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup, currentCID, cidAll);
        return result;
    }
    
    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()));
    }
    return result;
}

🎯 二、负载均衡算法 ​

2.1 一致性Hash分配 ​

AllocateMessageQueueConsistentHash：

java
@Override
public List<MessageQueue> allocate(String consumerGroup, String currentCID,
        List<MessageQueue> mqAll, List<String> cidAll) {
    if (currentCID == null || currentCID.length() < 1) {
        throw new IllegalArgumentException("currentCID is empty");
    }
    if (mqAll == null || mqAll.isEmpty()) {
        throw new IllegalArgumentException("mqAll is null or mqAll empty");
    }
    if (cidAll == null || cidAll.isEmpty()) {
        throw new IllegalArgumentException("cidAll is null or cidAll empty");
    }
    
    List<MessageQueue> result = new ArrayList<MessageQueue>();
    if (!cidAll.contains(currentCID)) {
        log.info("[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {}",
                consumerGroup, currentCID, cidAll);
        return result;
    }
    
    Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
    for (String cid : cidAll) {
        cidNodes.add(new ClientNode(cid));
    }
    
    final ConsistentHashRouter<ClientNode> router = 
            new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
    
    List<MessageQueue> results = new ArrayList<MessageQueue>();
    for (MessageQueue mq : mqAll) {
        ClientNode clientNode = router.routeNode(mq.toString());
        if (clientNode != null && currentCID.equals(clientNode.getKey())) {
            results.add(mq);
        }
    }
    
    return results;
}

📖 学习资源 ​

视频教程 ​

书籍推荐 ​

  • 《RocketMQ技术内幕》 - 丁威

网站资源 ​

✅ 学习检查清单 ​

  • [ ] 已理解队列分配策略
  • [ ] 已理解负载均衡算法
  • [ ] 已理解动态负载均衡
  • [ ] 已总结核心知识点

上一章节：事务消息实现
下一章节：顺序消息
返回目录：RocketMQ源码深度解析