RocketMQ的负载均衡策略

一、先搞懂核心概念：负载均衡的“基石”
在讲负载均衡之前，必须先明确两个核心组件的关系——Topic 与 Message Queue、消费者组（Consumer Group），这是理解负载均衡的前提。

1. Topic 与 Message Queue：消息的“存储单元”
RocketMQ 的 Topic 并非直接存储消息，而是通过消息队列（Message Queue，简称 MQ） 实现分片存储。

一个 Topic 会被划分为多个 Message Queue（数量可配置），每个 Queue 都是独立的消息存储单元，消息会分散写入不同 Queue。
例如 TopicA 可配置 4 个 Queue（Q1、Q2、Q3、Q4），生产者发送消息时，会按一定规则（如轮询、Hash）将消息分发到不同 Queue，实现消息的“分片存储”。
2. 消费者组：负载均衡的“执行单位”
负载均衡并非单个消费者的行为，而是以消费者组（Consumer Group） 为单位展开。

一个消费者组包含多个消费者实例（Consumer Instance），这些实例共同消费同一个 Topic 的消息。
RocketMQ 的规则是：同一个消费者组内，每个 Message Queue 只能被一个消费者实例消费。通过这种“Queue 独占”机制，避免了消息重复消费，同时实现了负载分担。
二、3 种常用负载均衡策略：怎么分 Queue？
RocketMQ 提供了多种内置负载均衡策略，默认策略可满足大多数场景，也支持自定义扩展。

1. 平均分配（AllocateMessageQueueAveragely）：默认且最常用
这是 RocketMQ 的默认策略，核心逻辑是“将 Queue 均匀分摊给每个消费者实例”，确保每个实例承担的 Queue 数量尽可能一致。

示例场景：

TopicA 有 4 个 Queue：Q1、Q2、Q3、Q4
消费者组 GroupA 有 2 个实例：Consumer1、Consumer2
分配结果：Consumer1 负责 Q1、Q2；Consumer2 负责 Q3、Q4
若 Queue 数量无法被消费者数量整除，会让前几个实例多承担 1 个 Queue。比如 5 个 Queue 分给 2 个实例，结果就是 Consumer1 负责 Q1、Q2、Q3，Consumer2 负责 Q4、Q5。

2. 按环形分配（AllocateMessageQueueByCircle）：顺序循环分配
这种策略会将 Queue 按顺序排成“环形”，再依次循环分配给每个消费者实例，适合需要按 Queue 顺序消费的场景。

示例场景：

TopicA 有 3 个 Queue：Q1、Q2、Q3
消费者组 GroupA 有 2 个实例：Consumer1、Consumer2
分配结果：Consumer1 负责 Q1、Q3；Consumer2 负责 Q2
分配逻辑类似“轮询”：先给 Consumer1 分 Q1，再给 Consumer2 分 Q2，接着回到 Consumer1 分 Q3，直到所有 Queue 分配完成。

3. 自定义分配策略：灵活适配业务
若内置策略无法满足需求，RocketMQ 允许通过实现 AllocateMessageQueueStrategy 接口，自定义 Queue 分配逻辑。

实现步骤：

实现 AllocateMessageQueueStrategy 接口，重写 allocate 方法，在方法中定义自己的分配规则（如按实例 IP Hash、按业务模块指定 Queue 等）。
消费者初始化时，通过 setAllocateMessageQueueStrategy 方法指定自定义策略。
代码示例（自定义策略骨架）：

// 1. 实现自定义策略接口
public class MyAllocateStrategy implements AllocateMessageQueueStrategy {
@Override
public List<MessageQueue> allocate(String consumerGroup, String currentCID, List<MessageQueue> mqAll, List<String> cidAll) {
// 自定义分配逻辑：例如按当前实例ID的Hash值分配Queue
List<MessageQueue> result = new ArrayList<>();
// 省略具体分配代码…
return result;
}

@Override
public String getName() {
return “MyAllocateStrategy”; // 策略名称
}
}

// 2. 消费者配置自定义策略
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(“GroupA”);
consumer.setAllocateMessageQueueStrategy(new MyAllocateStrategy()); // 指定自定义策略