RocketMQ是通过DefaultMQProducer
进行消息发送的,它实现了MQProducer
接口,MQProducer
接口中定义了消息发送的方法,方法主要分为三大类:
send
同步进行消息发送,向Broker发送消息之后等待响应结果;send
异步进行消息发送,向Broker发送消息之后立刻返回,当消息发送成功/失败之后触发回调函数;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)类似。
![](/i/l/?n=23&i=blog/2612945/202308/2612945-20230813224206827-8660896.png)
获取主题的发布信息(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为空,也就是首次发送消息时,处理逻辑如下:
- 对计数器增一;
- 根据计数器的值对消息队列列表的长度取余得到下标值
pos
,从队列列表中获取pos
位置的元素,以此达到轮询从消息队列列表中选择消息队列的目的; - 返回第2步中获取到的消息队列;
- 在调用获取消息队列的地方,会记录本次选择消息队列所在的BrokerName;
如果上一次选择的BrokerName不为空,表示上次发送消息时就发送给了此Broker,此时的处理逻辑与上面的不同点在第3步,通过从队列列表中获取pos
位置的元素之后,并没有直接把选取到的消息队列返回,而是再增加一个判断,判断当前选取到的Broker是否与上次选择的Broker名称一致,如果一致会继续循环,轮询选择下一个消息队列,如果不一致则直接返回:
- 对计数器增一;
- 根据计数器的值对消息队列列表的长度取余得到下标值
pos
,从队列列表中获取pos
位置的元素; - 对第2步获取到的消息队列进行判断:
- 如果本次选取到的队列与上次发送消息的Broker一致,回到第1步继续选择下一个队列,如果一直未选出满足要求的消息队列,则不作判断,使用上面的方式轮询选择一个队列返回;
- 如果本次选取到的队列与上次发送消息的Broker不一致,返回当前的队列;
- 在调用获取消息队列的地方,会记录本次选择消息队列所在的BrokerName;
总结
在未启用故障延迟机制时,从该消息所属的Topic下的所有消息队列集合中,轮询选择消息队列进行发送,如果上一次选择了某个Broker发送消息,本次将不会再选择这个Broker,当然如果最后仍未找到满足要求的消息队列,将会跳过这个判断,直接从队列中轮询获取消息队列返回。
开启故障延迟机制
在生产者进行发送消息的时候,无论消息是否发送成功与否都会记录向每个Broker的发送消息的条目信息FaultItem
,有一个失败条目表faultItemTable
,faultItemTable
记录了每个Broker对应的失败条目FaultItem
,FaultItem
中主要有以下信息:
- name:Broker的名称;
- currentLatency:延迟时间,可以理解为是本次向该Broker发送消息耗时时间:发送消息结束时间 - 消息发送开始时间;
- startTimestamp:规避故障时间,一般为
当前时间 + 不可用的持续时间
,不可用的持续时间有两种情况,分别为30000ms或者使用currentLatency延迟时间(也就是上次发送消息所用的时间),一般在出现异常的时候,会将不可用的持续时间设置为30000ms,消息正常发送的时候使用currentLatency
延迟时间。
设置规避故障时间主要是为了在某个时间段内规避某个Broker,假设向某个Broker发送失败/或者向此Broker发生消息的耗时比较长,生产者认为此Broker可能暂时处于异常状态/或者该时间段内此Broker的性能不高,在下次发送消息时尽量规避这个Broker,避免向此Broker上投递消息。
每次消息发送之后会更新该Broker的失败条目的处理逻辑如下:
- 根据Broker名称从
faultItemTable
获取对应的FaultItem
对象; - 如果上一步获取为空,说明之前没有记录过该Broker的信息,需要新建对应
FaultItem
对象,此时需要设置name、currentLatency延迟时间、startTimestamp规避故障时间; - 如果第1步中获取到该Broker对应的
FaultItem
对象,直接更新里面的currentLatency延迟时间、startTimestamp规避故障时间即可;
接下来看如何使用FaultItem
中记录的信息,来实现故障规避。
使用故障规避,需要启用故障延迟机制,此时从队列集合中选择消息队列的处理逻辑如下:
-
对计数器增一;
-
根据计数器的值对消息队列列表的长度取余得到下标值
pos
,从队列列表中获取pos
位置的元素,依旧轮询从消息队列列表中选择消息队列,这两步与未开启故障时逻辑一致; -
选择出消息队列之后,会获取该队列所在的Broker名称,上面说到,生产者每次与Broker通信发送消息时,会记录消息发发送情况,此时可以根据Broker的名称,从失败条目表
faultItemTable
中获取该Broker的FaultItem
,用来判断当前选择的消息队列是否可用,FaultItem
中有一个规避故障时间,来看两种情况:- 情况一:上次向此Broker发送消息失败,那么这个时间的值为
发送消息失败时的时间 + 30000ms
,判断当前时间有没有超过故障规避设置的时间,如果超过了当前选择的消息队列可用,那么就会返回当前选择的这个消息队列,如果未超过表示该Broker暂时不可用所以不能使用当前选择的消息队列,需要回到第1步继续选择下一个队列; - 情况二:上次向此Broker发送消息成功,那么这个时间的值为
发送消息失败时的时间 + 上次发送消息的耗时时间
,判断当前时间有没有超过故障规避设置的时间,这个依赖于上次发送消息的耗时时间
的长短,如果耗时比较长,可能还未超过规避时间,本次就不能选择向此Broker发送消息同样需要回到第1步选择下一个队列,如果耗时比较短,可能现在已经过了规避时间,那么就可以选择当前的消息队列返回;
- 情况一:上次向此Broker发送消息失败,那么这个时间的值为
-
如果进行到这一步,以上步骤没有选择到可用的消息队列,此时需要通过以下方式再次选择消息队列:
(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),然后返回当前选取的消息队列;
- 对比是否有超过规避时间的Broker(调用
-
如果经过第4步依旧未选出可用的消息队列,那么就跳过故障延迟机制,直接从该Topic的所有队列中轮询选择一个返回;
总结
故障延迟机制指的是在发送消息时记录每个Broker的耗时时间,如果某个Broker发生故障,但是生产者还未感知(NameServer 30s检测一次心跳,有可能Broker已经发生故障但未到检测时间,所以会有一定的延迟),用耗时时间做为一个故障规避时间(也可以是30000ms),此时消息会发送失败,在重试或者下次选择消息队列的时候,如果在规避时间内,可以避免再次选择到此Broker,以此达到故障规避的目的。
如果某个Topic所在的所有Broker都处于不可用状态,此时尽量选择延迟时间最短、规避时间最短(排序后的失败条目中靠前的元素)的Broker作为此次发送消息消息的Broker。
对应的相关源码可参考:
参考
标签:Topic,队列,Broker,发送,消息,选择,RocketMQ From: https://www.cnblogs.com/shanml/p/17627377.html