RocketMQ生产者如何规避故障Broker方式详解

this.updateFaultItem(mq.getBrokerName(), endTimestamp - beginTimestampPrev, true);

// 发送成功的话，isolation传false，失败isolation传true
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;
}
private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};

"broker-a": {
  // broker名称
  "name": "broker-a",
  "currentLatency": 发送消息消耗的时间，毫秒值之差,
  // 解除规避的时间，绝对时间
  "startTimestamp": 时间戳毫秒值
},
"broker-b": {
  // broker名称
  "name": "broker-b",
  "currentLatency": 发送消息消耗的时间，毫秒值之差,
  // 解除规避的时间，绝对时间
  "startTimestamp": 时间戳毫秒值
}

// 轮询的基本套路，一个自增变量
int index = tpInfo.getSendWhichQueue().incrementAndGet();
for (int i = 0; i < tpInfo.getMessageQueueList().size(); i++) {
     // 通过对队列数量取模，获取选定的Broker所在的位置
     int pos = Math.abs(index++) % tpInfo.getMessageQueueList().size();
     if (pos < 0)
         pos = 0;
     MessageQueue mq = tpInfo.getMessageQueueList().get(pos);
     // 判断Broker是否在规避时间内，如果不在规避时间内，就选择这个Broker，否则继续循环直至所有Broker都在规避时间内
     if (latencyFaultTolerance.isAvailable(mq.getBrokerName()))
         return mq;
}

// 把所有规避列表中的Broker按延迟高低排序，并从延迟低的Broker中选择一个
final String notBestBroker = latencyFaultTolerance.pickOneAtLeast();
int writeQueueNums = tpInfo.getQueueIdByBroker(notBestBroker);
// 判断该Broker是否允许写消息
if (writeQueueNums > 0) {
    final MessageQueue mq = tpInfo.selectOneMessageQueue();
    if (notBestBroker != null) {
        mq.setBrokerName(notBestBroker);
        mq.setQueueId(tpInfo.getSendWhichQueue().incrementAndGet() % writeQueueNums);
    }
    // 返回选中的Broker
    return mq;
}

return tpInfo.selectOneMessageQueue();

    public MessageQueue selectOneMessageQueue() {
        int index = this.sendWhichQueue.incrementAndGet();
        int pos = Math.abs(index) % this.messageQueueList.size();
        if (pos < 0)
            pos = 0;
        return this.messageQueueList.get(pos);
    }