队列选择
在HA的文章里大概讲了一下Producer如何为高可用贡献出一份力量的，当时只是说了遍历列表选择队列，然后选择一个，没有深入分析，这篇文章深入分析一下其源码，首先从发送消息选择队列的代码开始：
String lastBrokerName = null == mq ? null : mq.getBrokerName();
MessageQueue tmpmq = this.selectOneMessageQueue(lastBrokerName);
if (tmpmq != null) {
mq = tmpmq;
//....
如上，如果发送失败了，重试的时候lastBrokerName将不为空，进入到selectOneMessageQueue方法
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
return this.mqFaultStrategy.selectOneMessageQueue(tpInfo, lastBrokerName);
}
具体实现在mqFaultStrategy中
public MessageQueue selectOneMessageQueue(final TopicPublishInfo tpInfo, final String lastBrokerName) {
if (this.sendLatencyFaultEnable) {
try {
int index = tpInfo.getSendWhichQueue().getAndIncrement();
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
if (pos < 0)
pos = 0;
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
return mq;
}
}
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
if (writeQueueNums > 0) {
final MessageQueue mq = tpInfo.selectOneMessageQueue();
if (notBestBroker != null) {
mq.setBrokerName(notBestBroker);
mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
}
return mq;
} else {
latencyFaultTolerance.remove(notBestBroker);
}
} catch (Exception e) {
}
return tpInfo.selectOneMessageQueue();
}
return tpInfo.selectOneMessageQueue(lastBrokerName);
}
首先判断sendLatencyFaultEnable是否为true，来走不同的流程，默认为false，从意思上，那么先看下为false的情况，即tpInfo.selectOneMessageQueue(lastBrokerName)的代码实现：
public MessageQueue selectOneMessageQueue(final String lastBrokerName) {
// 如果为空，即第一次发生，未发生错误重试
// 直接轮询队列进行发送
if (lastBrokerName == null) {
return selectOneMessageQueue();
} else {
// 与selectOneMessageQueue类似，过滤的lastBrokerName的队列
int index = this.sendWhichQueue.getAndIncrement();
for (int i = 0; i < this.messageQueueList.size(); i++) {
int pos = Math.abs(index++) % this.messageQueueList.size();
if (pos < 0)
pos = 0;
MessageQueue mq = this.messageQueueList.get(pos);
if (!mq.getBrokerName().equals(lastBrokerName)) {
return mq;
}
}
return selectOneMessageQueue();
}
}
public MessageQueue selectOneMessageQueue() {
int index = this.sendWhichQueue.getAndIncrement();
int pos = Math.abs(index) % this.messageQueueList.size();
if (pos < 0)
pos = 0;
return this.messageQueueList.get(pos);
}
很简单的逻辑，总的来说都是轮询，只是一个有过滤失败的lastBrokerName，一个没有
接着看下sendLatencyFaultEnable开启的情况，sendLatencyFaultEnable这里分两部分来分析
- 第一部分代码如下：
int index = tpInfo.getSendWhichQueue().getAndIncrement();
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
if (pos < 0)
pos = 0;
MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
// 判断该Broker是否可用，不可用则进行第二部分的逻辑
if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {
// 非失败重试，直接返回到的队列
// 失败重试的情况，如果和选择的队列是上次重试是一样的，则返回
if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))
return mq;
}
}
- 第二部分代码如下：
//从容错信息中取一个Broker
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
if (writeQueueNums > 0) {// 有可写队列
// 往后取一个
final MessageQueue mq = tpInfo.selectOneMessageQueue();
if (notBestBroker != null) {
// 将取到的队列信息设置为取到的broker
mq.setBrokerName(notBestBroker);
// 队列重置
mq.setQueueId(tpInfo.getSendWhichQueue().getAndIncrement() % writeQueueNums);
}
return mq;
} else {
latencyFaultTolerance.remove(notBestBroker);
}
第一部分主要是选择一个可用的并且brokerName为lastBrokerName的队列，这里其实有点疑问，是失败的时候lastBrokerName才不为空，这时候为什么还会选择可用且brokerName为lastBrokerName的队列？这个猜测可能是觉得当前brokerName的上一次发送的队列失败了，可能下个队列会成功，加上当前延迟容错机制下的确保可用情况下，选择另外的队列。
假设没有找到对应的队列，只有一种情况
- 延迟容错机制觉得lastBrokerName这个broker不可用
那么将会进入第二部分代码，首先调用pickOneAtLeast获取一个broker，再调用selectOneMessageQueue获取一个队列，如果pickOneAtLeast取到的不为空，那么将队列信息替换，这里不知道为什么要取一个再替换，而且也导致了sendWhichQueue多增加了一次？
容错策略
上面讲了在选择队列的时候会判断容错策略的开关是否开启，如果开启了，则进入容错策略下的队列选择流程，上面一笔带过了部分代码，下面就分析一下该策略是如何判断broker是否可用，pickOneAtLeast代码的实现等几个地方，先看下isAvailable方法
public boolean isAvailable(final String name) {
final FaultItem faultItem = this.faultItemTable.get(name);
if (faultItem != null) {
return faultItem.isAvailable();
}
return true;
}
这里主要看分两部分
- faultItemTable是什么时候放进去的？
- FaultItem的isAvailable实现
isAvailable实现
public boolean isAvailable() {
return (System.currentTimeMillis() - startTimestamp) >= 0;
}
该方法很简单，判断当前时间是否大于startTimestamp，为什么只是判断一个时间就可以知道Broker是否可用？这需要分析第一部分代码
faultItemTable
通过查找faultItemTable使用的地方，找到updateFaultItem方法
public void updateFaultItem(final String name/*brokerName*/, final long currentLatency, final long notAvailableDuration) {
FaultItem old = this.faultItemTable.get(name);
if (null == old) {
final FaultItem faultItem = new FaultItem(name);
faultItem.setCurrentLatency(currentLatency);
faultItem.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
old = this.faultItemTable.putIfAbsent(name, faultItem);
if (old != null) {
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
}
} else {
old.setCurrentLatency(currentLatency);
old.setStartTimestamp(System.currentTimeMillis() + notAvailableDuration);
}
}
逻辑很简单，通过brokerName找到对应的FaultItem，startTimestamp=当前时间+notAvailableDuration，找到updateFaultItem使用的地方，看看notAvailableDuration是什么，找到MQFaultStrategy.updateFaultItem(String, long, boolean)方法
public void updateFaultItem(final String brokerName, final long currentLatency, boolean isolation) {
if (this.sendLatencyFaultEnable) {// 开启延迟容错功能
long duration = computeNotAvailableDuration(isolation ? 30000 : currentLatency);
this.latencyFaultTolerance.updateFaultItem(brokerName, currentLatency, duration);
}
}
private long computeNotAvailableDuration(final long currentLatency) {
for (int i = latencyMax.length - 1; i >= 0; i--) {
if (currentLatency >= latencyMax[i]) return this.notAvailableDuration[i];
}
return 0;
}
notAvailableDuration为notAvailableDuration数组某个位置的值，latencyMax和notAvailableDuration数组的值分别如下
| latencyMax | notAvailableDuration |
|---|---|
| 50L | 0L |
| 100L | 0L |
| 550L | 30000L |
| 1000L | 60000L |
| 2000L | 120000L |
| 3000L | 180000L |
| 15000L | 600000L |
即
- currentLatency如果大于等于50小于100，则notAvailableDuration为0
- currentLatency如果大于等于100小于550，则notAvailableDuration为0
- currentLatency如果大于等于550小于1000，则notAvailableDuration为300000
- ....以此类推
假设isolation传入true，那么notAvailableDuration将传入600000。
结合isAvailable方法，大概流程如下，RocketMQ为每个Broker预测了个可用时间(当前时间+notAvailableDuration)，当当前时间大于该时间，才代表Broker可用，而notAvailableDuration有6个级别和latencyMax的区间一一对应，根据传入的currentLatency去预测该Broker在什么时候可用
那么看下updateFaultItem使用的地方，看看currentLatency传入的是什么
// 1.
try {
beginTimestampPrev = System.currentTimeMillis();
sendResult = this.sendKernelImpl(msg, mq, communicationMode, sendCallback, topicPublishInfo, timeout);
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, false);
// 2.
} catch (xxException e) {
endTimestamp = System.currentTimeMillis();
this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);
}
currentLatency为发送消息的执行时间，根据执行时间来看落入哪个区间，在0~100的时间内notAvailableDuration都是0，都是可用的，大于该值后，可用的时间就会开始变大了，而在报错的时候isolation参数为true，那么该broker在600000毫秒后才可用(感觉时间有点久)
pickOneAtLeast
当真的出现600000毫秒后才可用的情况，在selectOneMessageQueue方法的第一部分代码就走不下去了，只能走到第二部分代码，先调用pickOneAtLeast方法获取一个broker
public String pickOneAtLeast() {
final Enumeration<FaultItem> elements = this.faultItemTable.elements();
List<FaultItem> tmpList = new LinkedList<FaultItem>();
// 将faultItemTable里的元素全放到list中
while (elements.hasMoreElements()) {
final FaultItem faultItem = elements.nextElement();
tmpList.add(faultItem);
}
if (!tmpList.isEmpty()) {
// 先打乱再排序
Collections.shuffle(tmpList);
Collections.sort(tmpList);
final int half = tmpList.size() / 2;
if (half <= 0) {// 只有一个元素的情况
return tmpList.get(0).getName();
} else {// 根据half取余
final int i = this.whichItemWorst.getAndIncrement() % half;
return tmpList.get(i).getName();
}
}
return null;
}
这里好像也没有什么特殊规律，用元素个数的一半的值来取余，而whichItemWorst也是类似sendWhichQueue的机制，感觉就是随机取了一个
上面就是整个队列选择以及容错的流程，总结一下：
- 在不开启容错的情况下，轮询队列进行发送，如果失败了，重试的时候过滤失败的Broker
- 如果开启了容错策略，会通过RocketMQ的预测机制来预测一个Broker是否可用
- 如果上次失败的Broker可用那么还是会选择该Broker的队列
- 如果上述情况失败，则随机选择一个进行发送
- 在发送消息的时候会记录一下调用的时间与是否报错，根据该时间去预测broker的可用时间
