咱们晓得RocketMQ的NameServer并非强统一而是最终一致性的,也就是客户端隔一段时间定时去获取Broker信息,如果Broker一段时间内呈现了故障,客户端并不能马上感应到,那RocketMQ如何做到音讯发送的高可用呢?大抵能够从上面三个方面来开展:

  • 重试机制
  • 顺次更换队列
  • 躲避己故障的Broker

一,重试机制
在发送音讯过程中有这样一段代码:

    int timesTotal = communicationMode == CommunicationMode.SYNC ? 1 + this.defaultMQProducer.getRetryTimesWhenSendFailed() : 1;    int times = 0;    for (; times < timesTotal; times++) {        ...    }

如果是同步发送的形式默认会重试3次,重试次数也能够通过retryTimesWhenSendFailed进行配置。

二,顺次更换队列
在selectOneMessageQueue办法在里,会从sendWhichQueue获得上一次应用过的队列的索引,这个sendWhichQueue是一个ThreadLocalIndex类型,外面有一个ThreadLocal,队列索引就是存储在这个ThreadLocal里,获得索引后会有一个自增的动作,而后再依据新的索引获取新的队列。举个例子:有一个名为broker-a的broker,外面有队列q1,q2,q3,q4,如果上一次应用的是队列q1那么索引自增后会抉择q2作为新的发送队列。通过这样的形式会让这4个队列的负载尽量保持一致。代码如下:

                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;                    //获取一个队列,队列里记录着所属broker                    MessageQueue mq = tpInfo.getMessageQueueList().get(pos);                    //判断 broker 是否生效                    if (latencyFaultTolerance.isAvailable(mq.getBrokerName())) {                        //第一次获取会是空;获取的broker是无效的且还是上一次获取应用过的                        if (null == lastBrokerName || mq.getBrokerName().equals(lastBrokerName))                            return mq;                    }                }

躲避己故障的Broker

如果上次应用的Broker呈现了故障或延时超过了肯定的范畴,RocketMQ是如何躲避的呢?
每次发送完音讯后RocketMQ都应用ConcurrentHashMap记录该broker与发送音讯所花工夫的对应关系,在下次发送音讯之前会依据这个工夫判断一下队列所在Broker是否能够应用,如果是可用的并且也是上次应用的Broker则间接应用,这里咱们重点看一下获取的Broker不可应用的状况。咱们假如有broker-a,broker-b两个broker,每个队列里都有q1,q2,q3,q4这四个队列:

在发送之前 sendWhichQueue 该值为 broker-a 的 q1,如果因为此时 broker-a 的突发流量异样大导致音讯发送失败,会触发重试,依照轮循机制,下一个抉择的队列为 broker-a 的 q2 队列,此次音讯发送大概率还是会失败,即只管会重试 2 次,但都是发送给同一个 Broker 解决,此过程会显得不那么靠谱,即大概率还是会失败,那这样重试的其实是没多大意义的。

RocketMQ 为了解决该问题,引入了故障躲避机制,在音讯重试的时候,会尽量躲避上一次发送的 Broker,回到上述示例,当音讯发往 broker-a q1 队列时返回发送失败,那重试的时候会先排除 broker-a 中所有队列,即这次会抉择 broker-b q1 队列,增大音讯发送的成功率。RocketMQ 提供了两种躲避策略,该参数由 sendLatencyFaultEnable 管制,用户可干涉,示意是否开启提早躲避机制,默认为不开启。躲避broker-a的代码如下:

public String pickOneAtLeast() {        final Enumeration<FaultItem> elements = this.faultItemTable.elements();        List<FaultItem> tmpList = new LinkedList<FaultItem>();        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 {                final int i = this.whichItemWorst.getAndIncrement() % half;                return tmpList.get(i).getName();            }        }        return null;    }

开启提早躲避机制,一旦音讯发送失败会将 broker-a “乐观”地认为在接下来的一段时间内该 Broker 不可用,在为将来某一段时间内所有的客户端不会向该 Broker 发送音讯。这个延迟时间就是通过 notAvailableDuration、latencyMax 独特计算的,就首先先计算本次音讯发送失败所耗的时延,而后对应 latencyMax 中哪个区间,即计算在 latencyMax 的下标,而后返回 notAvailableDuration 同一个下标对应的提早值.例如,如果上次申请的latency超过550Lms,就退却3000Lms;超过1000L,就退却60000L;

private long[] latencyMax = {50L, 100L, 550L, 1000L, 2000L, 3000L, 15000L};private long[] notAvailableDuration = {0L, 0L, 30000L, 60000L, 120000L, 180000L, 600000L};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;    }

RocketMQ就是通过失败重试,顺次更换队列还有躲避有故障的broker这三个方面来保障音讯发送的高可用的。