RocketMQ底层原理

每一个不同的消费组都会接收到这个消息,区别是可以选择是一个消费组就一台机器接收还是消费组中所有的机器接收

可以通过一个参数设置:consumer.setMessageModel(MessageModel,BROADCASTING);
默认情况下都是集群模式,一个消费组获取到一条消息,交给组内的一台机器去处理
广播模式就是消费组内的每台机器都可以获取到这条消息

对于一个topic上的多个messagequeue,是平均分配给消费组中的多个机器的

push模式的本质也是去broker上拉取消息,意思是Broker会尽可能实时的把新消息交给消费者进行处理
push模式的实现思路是当消费者发送请求到broker去拉取消息的时候,如果有新的消息可以消费,会立马返回一批消息到消费者机器处理,处理完之后会立刻发送请求到broker去拉取下一批消息.消息的时效性非常好,看起来就跟broker不断推送消息一样.push模式下有一个请求挂起和长轮询的机制,就是当消费者去拉取消息的时候,broker发现没有消息可以给,就会暂时让请求线程挂起,默认是15秒,然后这段时间有个后台线程每隔一段时间就去检查是否有新的消息给你,另外如果在这个挂起过程,有新的消息到达了会主动唤醒挂起的线程,然后把消息返回给消费者

基于ConsumeQueue读取消息在CommitLog中的物理位置,也就是offset

根据偏移量从commitLog中读取消息数据

回调消费者服务器注册的监听函数来处理

提交消息处理进度

broker存储消息消费的consumerOffset

消费组内机器宕机

消费组进行机器的扩容

broker写入消息到commitLog文件中,是先写入page cache中,然后异步刷盘存储到磁盘的commitLog文件中

ConsumeQueue会被大量的消费者频繁的读取,这个读取的操作如果是从磁盘读,会影响到消息的写入和写出,进而影响到消息的吞吐量,所以实际上rocketmq的consumequeue是基于os cache进行优化的,os 自己有一个优化机制,就是读取一个磁盘文件的时候,会自动把磁盘文件的一些数据缓存到os cache中,而且本身consumequeue文件也不大,5.72MB,所以是完全可以被os缓存在cache中,也就是说对consumequeue的读取是内存级别的,这个性能就有了保障

消费者读取消息,有可能从磁盘读,也有可能从os cache读,取决于消费消息的进度

根据消费者消费进度,也就是未消费的消息数量和os cache中最大能够存储的消息数量的对比
如果消息还在os cache中,那就从os cache中读,如果不在,那就从磁盘中读

当消费者去master拉取消息,master发现消费者有8w的消息未消费,而当前master可以存储在os cache中的消息只有5W,也就是说,接下来消费者会有大量的磁盘读取的操作,这个时候会指示消费者从slave读取消息,其实本质上是对比当前没有拉取消息的数量和大小,以及最多可以存放在os cache内存里的消息的大小,如果没拉取的消息超过了最大能使用的内存的量,那么说明后续会频繁的从磁盘加载数据,此时就让消费者去slave broker拉取消息了

创建topic时需要设置messageQueue,本质上是数据分片机制,通过MessageQueue把一个topic主题的消息拆分了多个部分

Messagequeue均匀的分散到broker集群

broker会把自身的topic信息,以及message信息发送给nameServer上,然后生产者写入消息之前会从nameserver上拉取路由信息,知道哪些MessageQueue在哪台机器上

在Producer中开启一个开关,就是sendLatencyFaultEnable,一旦打开这个开关,那就会有一个自动容错机制,比如,如果某次访问一个Broker发现网络延迟有500ms,然后还无法访问,那么接下来就会自动回避访问这个Broker,比如接下来的3000ms,不会访问

各种mq,不只是让你写入消息和获取消息那么简单,本身最重要的就是提供强大的数据存储能力,可以把亿万级的海量消息存储在自己的服务器的磁盘上,这样的话各种不同的系统从mq中消费消息的时候,才可以从mq服务器上获取自己需要的消息.
所以Broker数据存储实际上才是mq最重要的环境,他决定了生产者消息写入的吞吐量,决定了消息不能丢失,决定了消费者获取消息的吞吐量

broker收到消息后会顺序追加写到CommitLog文件中,注意这个时候并不会直接写入磁盘,而是写入到os的pageCache内存缓存中,然后在某一个时间,os后台线程把数据刷盘写入磁盘,每个commitlog的大小不超过1GB,这也是方便内存映射可以直接转载到内存中

消息写入到commitlog的时候也会同时将这条消息在commitLog中的物理位置也就是一个文件偏移量,就是一个offset,写入到messagequeue所属的consumequeue文件中去,所以实际上consumequeue中存储的每条数据都是commitlog文件中一个消息的引用,当然consumequeue中存储的每条数据不只是消息在commitlog中的offset偏移量,还包含了消息的长度,以及tag hashcode,一条数据是20字节,每个consumequeue文件保存30W调数据,大概每个文件大小是5.72MB

文件顺序追加写入

基于os cache写入

os后台线程异步刷盘

Dledger会把Broker的CommitLog文件替换为自己管理的CommitLog文件,
然后Broker还是可以基于Dledger的CommitLog去构建机器上的ConsumeQueue磁盘文件

Dledger基于Raft算法协议进行选举

Raft协议的随机休眠机制

Leader可以写入,Follower不可以写入

uncommitted消息同步给follower

过半follower返回ack即可返回

Dledger在本地执行commit

Dledger将commit消息发送给follower

剩余Follower重新选举Leader

Leader自动热切换

Leader自动完成数据恢复

新的leader继续执行写入