之前看rocketmq,然后在想一个问题,就是一主一从的集群结构中,如果master宕机了,consumer这边是怎么选择的,按照官方说明中,master挂了,但是slave得消息仍然可以被消费到。原因是master和slave是一直有连接的,所以master上面的消息是可以及时同步到slave,但是终究是有一部分留在master(异步复制)的时候。
那么问题来了,consumer是怎么从master上面切换到slave上继续消费消息呢?首先明确一点,master宕机,就意味着这个broker不再写入,但是因为slave还在,所以还可以继续读。所以我们看一下consumer是怎么选择的?
从Pullconsumer 进去,调用pullBlockIfNotFound方法,一直进去,最后到:DefaultMQPullConsumerImpl的pullSyncImpl 方法,最后看到这段:
PullResult pullResult =this.pullAPIWrapper.pullKernelImpl(mq,suçbscriptionData.getSubString(),0L,offset,maxNums,sysFlag,0,this.defaultMQPullConsumer.getBrokerSuspendMaxTimeMillis(),timeoutMillis,CommunicationMode.SYNC,null);
进去看一下实现:
FindBrokerResult findBrokerResult =
this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(),
this.recalculatePullFromWhichNode(mq),false);
其实这段也很好理解,就是先根据订阅的关系找到broker。是根据broker的名字和id共同起作用的。broker的名字肯定能够拿到一堆broker的id,一般都是一主多从,那这个怎么选呢?
进去看看就知道了:
HashMap map =this.brokerAddrTable.get(brokerName);
if(map !=null&& !map.isEmpty()) {
brokerAddr = map.get(brokerId);
slave = brokerId != MixAll.MASTER_ID;
found = brokerAddr !=null;
if(!found && !onlyThisBroker) {
Entry entry = map.entrySet().iterator().next();
brokerAddr = entry.getValue();
slave = entry.getKey() != MixAll.MASTER_ID;
found =true;
}
}
先拿到broker的缓存,其实就是存在本地的hashmap,然后根据broker的id查找,如果找到了,判断下是不是slave角色返回,找不到的情况下就根据那拿到的列表迭代一个出来,考虑到时无序的,所以就可以理解为随机拿一个出来了,再判定角色。
所以传进来的brokerId非常重要,如果这台机器没有宕机的情况下,就是返回这个broker的地址了,否则就是从剩下的机器进行随机一个。
那传进去的brokerId是怎么产生的呢?
public longrecalculatePullFromWhichNode(finalMessageQueue mq) {
if(this.isConnectBrokerByUser()) {
return this.defaultBrokerId;
}
AtomicLong suggest =this.pullFromWhichNodeTable.get(mq);
if(suggest !=null) {
returnsuggest.get();
}
returnMixAll.MASTER_ID;
}
传入的mq是负载均衡服务分配给当前consumer消费的队列,它必然是属于一个brokername唯一的拓扑结构中,即一主多从的几台机器中,从哪个机器选就很重要了,因为都可以选的。
1.先看isConnectBrokerByUser 是否设置,如果设置,返回默认的,即0.
2.看缓存中是否已经存了建议值,如果存了,直接返回
3.返回master的,即0.
那到这里,我们看一下怎么解释一下宕机时主从切换过程,consumer时如何从主上面切换到从上面的?
1.一开始时正常的,因为没有缓存,也没有特别设置,所以,进入3 返回master。
2.master 写入缓存,后面都读取到缓存,在上面的步骤2中返回。
master宕机了,然后nameserver中不再收到心跳,然后master机器剔除掉。所以consumer虽然选到了master,但是因为在地址中找不到broker'id=0的数据,于是进入随机过程,然后这样就切到了slave,然后slave写入到缓存。后面一直读到缓存中的slave。
问题来了:master起来后,建议值的缓存也没有更新,那怎么切回到master,毕竟我们是因为宕机产生地址找不到的时候,才能完成切换的,这解释不通。
后来debug了一发,发现这个suggest值并不是consumer端决定的,而是broker决定的。啥意思,即使你是拉去slave上面的数据,slave上面返回的结果中的suggest值也可能是0,然后0就写进缓存中,下一次,你还是优先访问master,然后master没有地址,访问slave。
这个意思就是,consumer端是根据缓存中的suggest值优先选机器。但是呢这个suggest是通过broker传回来的。所以即使是访问slave,传回来的suggest值仍然是master,只不过客户端没有master的映射关系,所以继续访问slave。这就能解释为什么master从宕机起来后,consumer能够切回master,因为地址映射表得到更新了,nameserver中有了master的信息了。
那么重点来了:
什么时候,broker返回的建议值是0?
什么时候,broker返回的建议值是其他值?
1.如果master中的堆积信息过多,默认返回consumerslow配置,默认是1.(所以机器的brokerId真的不能乱用),这个时候就切到slave了。
2.
未完待续~