【RocketMQ】消息的发送
阅读原文时间:2023年08月14日阅读:3

RocketMQ是通过DefaultMQProducer进行消息发送的,它实现了MQProducer接口,MQProducer接口中定义了消息发送的方法,方法主要分为三大类:

  1. send同步进行消息发送,向Broker发送消息之后等待响应结果;

  2. send异步进行消息发送,向Broker发送消息之后立刻返回,当消息发送成功/失败之后触发回调函数;

  3. sendOneway单向发送,也是异步消息发送,向Broker发送消息之后立刻返回,但是没有回调函数;

    public interface MQProducer extends MQAdmin {

    // 同步发送消息
    SendResult send(final Message msg) throws MQClientException, RemotingException, MQBrokerException,
        InterruptedException;
    
    // 异步发送消息,SendCallback为回调函数
    void send(final Message msg, final SendCallback sendCallback) throws MQClientException,
        RemotingException, InterruptedException;
    
    // 异步发送消息,没有回调函数
    void sendOneway(final Message msg) throws MQClientException, RemotingException,
        InterruptedException;

    }

接下来以同步发送为例,看下生产者发送消息的过程。

Topic主题

一般会为同一业务类型设定一个Topic,将不同的业务类型的数据放到不同的Topic中管理,不过主题只是一个逻辑概念,并不是实际的消息容器。

主题内部由多个消息队列(MessageQueue)组成,消息队列是消息存储的实际容器,消息队列与Kafka的分区(Partition)类似。

获取主题的发布信息(TopicPublishInfo)

Broker在启动时会向NameServer发送注册信息并定时向NameServer发送心跳包,Broker向NameServer注册信息中包括了该Broker的IP、Name以及Topic的配置信息,

生产者和消费者默认每30s从NameServer更新一次路由信息,可以知道消息所在的Topic分布在哪些Broker上。

生产者有一个主题路由信息表topicPublishInfoTable,缓存从NameServer拉取到的路由信息,它是ConcurrentMap类型的,KEY为topic主题名称, value为该Topic的发布信息,是TopicPublishInfo类型。

当生产者向Broker发送消息之前,首先需要知道消息所属的Topic的路由信息,有了Topic的路由信息才能知道Topic分布在哪个Broker上,生产者往哪个Broker上发,而topicPublishInfoTable中记录了每个主题的相关信息,可以从topicPublishInfoTable中查找Topic的路由信息。

如果从topicPublishInfoTable中查找成功,就可以继续后续的步骤,如果查找失败,此时生产者需要从NameServer中查询该Topic的路由信息:

  • 如果查询成功,会判断路由信息是否发生了变化,如果发生变化,生产者会更新本地缓存的该Topic的路由信息;
  • 如果依旧未查询到,它会有一个默认的主题,会使用这个默认的主题进行消息发送;

选取消息队列

前面知道,一个Topic一般由多个消息队列组成,所以主题的发布信息数据TopicPublishInfo获取到之后,需要从中选取一个消息队列,然后获取此消息队列所属的Broker,与Broker通信将消息投递到对应的消息队列中。

未启用故障延迟机制

在每个Topic内部,设置了一个计数器sendWhichQueue用于轮询从消息队列集合中选取队列。

在未启用故障延迟机制的时候,如果上一次选择的BrokerName为空,也就是首次发送消息时,处理逻辑如下:

  1. 对计数器增一;
  2. 根据计数器的值对消息队列列表的长度取余得到下标值pos,从队列列表中获取pos位置的元素,以此达到轮询从消息队列列表中选择消息队列的目的;
  3. 返回第2步中获取到的消息队列;
  4. 在调用获取消息队列的地方,会记录本次选择消息队列所在的BrokerName;

如果上一次选择的BrokerName不为空,表示上次发送消息时就发送给了此Broker,此时的处理逻辑与上面的不同点在第3步,通过从队列列表中获取pos位置的元素之后,并没有直接把选取到的消息队列返回,而是再增加一个判断,判断当前选取到的Broker是否与上次选择的Broker名称一致,如果一致会继续循环,轮询选择下一个消息队列,如果不一致则直接返回:

  1. 对计数器增一;
  2. 根据计数器的值对消息队列列表的长度取余得到下标值pos,从队列列表中获取pos位置的元素;
  3. 对第2步获取到的消息队列进行判断:
    • 如果本次选取到的队列与上次发送消息的Broker一致,回到第1步继续选择下一个队列,如果一直未选出满足要求的消息队列,则不作判断,使用上面的方式轮询选择一个队列返回;
    • 如果本次选取到的队列与上次发送消息的Broker不一致,返回当前的队列;
  4. 在调用获取消息队列的地方,会记录本次选择消息队列所在的BrokerName;

总结

在未启用故障延迟机制时,从该消息所属的Topic下的所有消息队列集合中,轮询选择消息队列进行发送,如果上一次选择了某个Broker发送消息,本次将不会再选择这个Broker,当然如果最后仍未找到满足要求的消息队列,将会跳过这个判断,直接从队列中轮询获取消息队列返回。

开启故障延迟机制

在生产者进行发送消息的时候,无论消息是否发送成功与否都会记录向每个Broker的发送消息的条目信息FaultItem,有一个失败条目表faultItemTablefaultItemTable记录了每个Broker对应的失败条目FaultItemFaultItem中主要有以下信息:

  • name:Broker的名称;
  • currentLatency:延迟时间,可以理解为是本次向该Broker发送消息耗时时间:发送消息结束时间 - 消息发送开始时间;
  • startTimestamp:规避故障时间,一般为当前时间 + 不可用的持续时间,不可用的持续时间有两种情况,分别为30000ms或者使用currentLatency延迟时间(也就是上次发送消息所用的时间),一般在出现异常的时候,会将不可用的持续时间设置为30000ms,消息正常发送的时候使用currentLatency延迟时间。

设置规避故障时间主要是为了在某个时间段内规避某个Broker,假设向某个Broker发送失败/或者向此Broker发生消息的耗时比较长,生产者认为此Broker可能暂时处于异常状态/或者该时间段内此Broker的性能不高,在下次发送消息时尽量规避这个Broker,避免向此Broker上投递消息。

每次消息发送之后会更新该Broker的失败条目的处理逻辑如下:

  1. 根据Broker名称从faultItemTable获取对应的FaultItem对象;
  2. 如果上一步获取为空,说明之前没有记录过该Broker的信息,需要新建对应FaultItem对象,此时需要设置name、currentLatency延迟时间、startTimestamp规避故障时间;
  3. 如果第1步中获取到该Broker对应的FaultItem对象,直接更新里面的currentLatency延迟时间、startTimestamp规避故障时间即可;

接下来看如何使用FaultItem中记录的信息,来实现故障规避。

使用故障规避,需要启用故障延迟机制,此时从队列集合中选择消息队列的处理逻辑如下:

  1. 对计数器增一;

  2. 根据计数器的值对消息队列列表的长度取余得到下标值pos,从队列列表中获取pos位置的元素,依旧轮询从消息队列列表中选择消息队列,这两步与未开启故障时逻辑一致;

  3. 选择出消息队列之后,会获取该队列所在的Broker名称,上面说到,生产者每次与Broker通信发送消息时,会记录消息发发送情况,此时可以根据Broker的名称,从失败条目表faultItemTable中获取该Broker的FaultItem,用来判断当前选择的消息队列是否可用,FaultItem中有一个规避故障时间,来看两种情况:

    • 情况一:上次向此Broker发送消息失败,那么这个时间的值为发送消息失败时的时间 + 30000ms,判断当前时间有没有超过故障规避设置的时间,如果超过了当前选择的消息队列可用,那么就会返回当前选择的这个消息队列,如果未超过表示该Broker暂时不可用所以不能使用当前选择的消息队列,需要回到第1步继续选择下一个队列;
    • 情况二:上次向此Broker发送消息成功,那么这个时间的值为发送消息失败时的时间 + 上次发送消息的耗时时间,判断当前时间有没有超过故障规避设置的时间,这个依赖于上次发送消息的耗时时间的长短,如果耗时比较长,可能还未超过规避时间,本次就不能选择向此Broker发送消息同样需要回到第1步选择下一个队列,如果耗时比较短,可能现在已经过了规避时间,那么就可以选择当前的消息队列返回;
  4. 如果进行到这一步,以上步骤没有选择到可用的消息队列,此时需要通过以下方式再次选择消息队列:

    (1)遍历faultItemTable失败条目表,将每一个Broker对应的FaultItem加入一个LinkedList链表;

    (2)对链表进行排序,FaultItem实现Comparable就是为了在这里进行排序,值小的排在链表前面,值的大小判断规则如下:

    • 对比是否有超过规避时间的Broker(调用isAvailable可以判断),如果有表示值比较小,会排在前面,之后被优先选择,如果所有的Broker都为超过规避时间,进入下一个对比条件;
    • 对比currentLatency的值,值越小排序的时候越靠前,也就是尽量选择发送消息耗时短的那个Broker,如果值相等进入下一个对比条件;
    • 对比startTimestamp的值,同样值越小排序的时候越靠前,尽量选择规避时间较短的那个Broker;

    (3)经过以上的规则进行排序后,会根据链表的总大小,计算一个中间值:

    • 如果half值小于等于0,取链表中的第一个元素;
    • 如果half值大于0,从前half个元素中轮询选择元素;

    (4)在链表中越靠前的元素,表示发送消息的延迟越低,在选择时优先级就越高,如果half值小于等于0的时候,取链表中的第一个元素,half值大于0的时候,处于链表前half个的Broker,延迟都是相对较低的,此时轮询从前haft个Broker中选择一个Broker,总之经过这么多处理就是为了选择一个延迟相对较低的Broker

    (5)获取上一步选取到的那个Broker,获取Broker可写的队列数量:

    • 如果数量小于0表示该Broker不可用,需要移除然后进入下一步;
    • 如果数量大于0,表示该Broker可用,然后重新轮询从消息队列列表中选取一个队列,将本次选取到的消息队列所属的Broker设置为第(4)步中选取到的那个Broker,也就是将这个消息队列及Topic重置到新的Broker中(认为原本所属的Broker不可用,需要设置一个新的Broker),然后返回当前选取的消息队列;
  5. 如果经过第4步依旧未选出可用的消息队列,那么就跳过故障延迟机制,直接从该Topic的所有队列中轮询选择一个返回;

总结

故障延迟机制指的是在发送消息时记录每个Broker的耗时时间,如果某个Broker发生故障,但是生产者还未感知(NameServer 30s检测一次心跳,有可能Broker已经发生故障但未到检测时间,所以会有一定的延迟),用耗时时间做为一个故障规避时间(也可以是30000ms),此时消息会发送失败,在重试或者下次选择消息队列的时候,如果在规避时间内,可以避免再次选择到此Broker,以此达到故障规避的目的。

如果某个Topic所在的所有Broker都处于不可用状态,此时尽量选择延迟时间最短、规避时间最短(排序后的失败条目中靠前的元素)的Broker作为此次发送消息消息的Broker。

对应的相关源码可参考:

【RocketMQ】【源码】MQ消息发送

参考

RocketMQ官方文档