Consumer的负载均衡
RocketMQ在消费端的负载均衡如下图所示，各个partition均匀分布在各个consumer上，如下图所示：
image
所有consumer依次消费每一个partition，如果partition数量不是consumer的整数倍，则排在前面的consumer会消费更多的partition，下面可以看看consumer的实现。
Rebalance的实现
RocketMQ的consumer负载均衡依赖于RebalanceImpl类，以push的方式为例，在DefaultMQPushConsumerImpl为例，其中包含RebalancePushImpl：
image
RebalanceImpl负责消费端的负载均衡，其中的doRebalance方法：
image
我们再进入到rebalanceImpl的doRebalance方法，其中调用了rebalanceByTopic，我们看看rebalanceByTopic中的逻辑：
image
image
可以看到，其主体逻辑比较简单：
- 先获取topic下的MessageQueue，一个MessageQueue实际上就是一个partition
- 然后获取当前topic和group的client id，即当前group中消费此topic的客户端
- 随后对partition和client id做排序
- 然后调用strategy获取当前客户端需要消费的partition
- 最后更新订阅
因此，负载均衡的主体算法在AllocateMessageQueueStrategy中实现，通过DefaultMQPushConsumer的默认构造器我们可以看到，默认使用的AllocateMessageQueueStrategy是AllocateMessageQueueAveragely实现类：
image
找到具体的实现类后，我们可以看到默认使用的负载均衡算法：
image
公式写的非常绕，代几个数进去算一下就知道，默认情况下，rocketmq使用的是连续分配的方式，示意图如下：
image
AllocateMessageQueueStrategy提供了多个实现：
image
- AllocateMessageQueueAveragely是前面讲的默认方式
- AllocateMessageQueueAveragelyByCircle则是本文最前面的示意图，每个消费者依次消费一个partition
- AllocateMessageQueueConsistentHash使用的是一致性hash算法
- AllocateMachineRoomNearby是通过就近元则，离的近的消费
- AllocateMessageQueueByConfig是通过配置的方式
在不同的情况下，我们可以选择使用不同的负载均衡实现方式。
对于特定场景，甚至可以自己实现负载均衡策略，比如我们的应用需要消费非常多个topic，而每个topic的partition不一定刚才都是机器 数量的整数倍，这个时候，按照rocketmq提供的负载均衡策略，排在前面的consumer消费的partition数量会多于后面的consumer，当topic非常多时，这就导致排在前面的consumer消费的partition比后面的consumer要多很多，造成集群中不同机器的水位相差非常大，这种场景下就知道自己实现负载均衡策略
何时重新Rebalace
这里先要介绍一个类MQClientInstance，此类在DefaultMQPushConsumerImpl的start方法中有如下代码：
image
这里的mQClientFactory的实现类实际上就是一个MQClientInstance，进入到MQClientInstance类的构造器中，我们可以看到它创建了一个RebalanceService对象，代码如下：
image
我们一级级的看下去，在RebalanceService的run方法中，可以看到，默认每20s调一次doRebalance:
image
而在父类ServiceThread中，我们可以看到run方法的调用方式，实际上是创建了一个线程：
image
因此，当一个consumer出现宕机后，默认最多20s，其它机器将重新消费已宕机的机器消费的partition，同样当有新的consumer连接上后，20s内也会完成rebalance使得新的consumer有机会消费partition
网友评论