【RocketMq系列-02】RocketMq的架构解析和高性能设计

RocketMq系列整体栏目

---

内容	链接地址
【一】RocketMq安装和基本概念	https://zhenghuisheng.blog.csdn.net/article/details/134486709
【二】RocketMq的架构解析和高性能设计/font>	https://zhenghuisheng.blog.csdn.net/article/details/134559514

---

一，RocketMq的架构解析和高性能设计

在rocketMq中，其整体架构如下，在RocketMqServer中，主要有NameServer，Broker，MessageQueue，Message等组件，并且存在Topic这种逻辑组件，表示一种主题

NameServer是topic的注册中心，NameServer会和topic建立长连接，将broker的信息通过topic注册到NameServer中，然后生产者和消费者都会先通过这个NameServer获取相关信息，再和对应的broker建立长连接。

在微服务中，有Nacos，zookeeper等作为注册中心：

但是zk很明显不适合作为这种高可用的注册的这中心，因为内部可能会因为选举出现脑裂问题，并且因为这个问题可能会导致整个服务出现一定时间的不可用的问题，而rocketmq主要就是高吞吐量，低延迟的特性，因此不可能去选择zk作为注册中心的；

而nacos和eureka也不适合作为rocketmq的注册中心，如nacos中会记录很多信息，如心跳信息，端口，host等信息，而Nameserver中只需要记录这个Broker的信息，如果使用nacos来做的话，有点大材小用了。并且如果引用nacos，还要考虑版本冲突这些，做一些适配器等，相对来说是更加复杂的

在topic中的Consumer配置中，每个topic都会对应一个或者多个消费者组，topic主题和消费者组是多对多的关系，一个consumer消费者组，代表的是一组逻辑相同的消费者，一个message消息，只能被消费者组中的一个消费者消费，这个和kafka中的消息消费是一样的

上面提到了消费者组的概念，在生产者中，也有生产者组。在事务机制中，当生产者给broker发送数据之后，broker需要给生产者一个数据回调，那么就需要指定生产者名字，那么此时生产者组就能发挥其作用

生产者producer在本地会有一个缓存存储Nameserver中存储的broker，在往broker投递之前，会向注册中心中发起一个请求判断是否需要拉取最新的配置，然后再往对应的broker发送数据

2，rocketmq底层原理

2.1，事务的底层实现

rocketmq的事务实现，相当于一个简单的分布式事务，主要是保证生产者本地事务和发送到broker事务的原子性。而broker到consumer端是一定可以保证消息消费成功的，如果一个消费者失败，那么可以往别的消费者里面推送，如果最终依旧失败，那么可以先重试，最后加入到死信队列里面

事务消息的底层实现如下图，首先生产者会发送一个half消息给Broker，Broker在接收到这个half消息之后，就会向broker返回一个确认的标志，然后事务的发送者就会执行本地事务，通过这个execute去执行本地事务。如果本地事务执行成功，那么生产者会返回一个提在交的状态给Broker，随后Broker将消息投递到消费者中；如果是回滚状态，那么消息会直接丢掉；如果是在4的时候，本地事务需要的时间过长，那么本地会先返回一个unknow的未知状态，然后broker会等一段时间，随后再回生产者中定时回查，消息生产者会去检查事务，默认是回查15次，如果是15次之后检查还是没有完成，那么消息就会直接丢弃掉

half消息有点类似于建立tcp连接，主要是做为一种嗅探机制，判断当前broker服务是否正常，如果broker服务挂了，那么连本地事务，也可以直接不执行了。

如一个订单场景，30s检查一次是否支付，那么就可以直接通过这种事务去实现，通过execute方法去执行本地事务，然后通过这个check的方式去银行进行对账。如果最终超时，那么最终将消息放入到死信队列中，在私信队列中写对应的逻辑，如将库存加回等。

2.2，如何保证消息不丢失

在mq中，消息丢失主要有四个地方，分别是生产者到broker、broker到消费者，broker的master到slave以及操作系统自身的缓存。

• 生产者到broker的解决方案可以如下：可以选择最简单的同步+多次试错的方式，或者可以直接选择事务消息
• broker到消费者之间：消费者本身具有重试功能，消费者不应答就会往别的消费者投递
• 操作系统主要是因为数据在缓存，如果出现断电而未来得及刷盘导致，因此应该采用同步刷盘解决
• broker到的master到slave之间：也可以采用同步的方式，来一条消息就往slave写入，或者通过Dledger集群

操作系统和主从之间保证消息不丢失，主要是通过同步的方式解决，但是在保证安全的情况下，会在一定的程度上影响吞吐量和性能

2.3，rocketmq积压问题

在rocketmq中，其处理数据积压问题时比其他mq的能力强的，如果出现积压，那么可以直接通过控制台上面的topic，通过内部的代理者位点和消费者位点所产生的差值查看，如果差值为0，则表示有消息积压未处理。

在rocketmq内部，一个MessageQueue队列的消息只能由一个消费者组中的一个消费者去消费，其底层实现和kafka是一样的，因此如果出现消息积压，那么首先可以查看消费者组中的消费者个数和队列的个数是否相同，如果消费者个数小于队列的个数，那么可以增加消费者个数，直到和队列的个数一致，如默认队列的个数为4，那么将消费者组中的消费者个数设置成4

当然，消费者个数调大是没有用的，因为最大只能和topic中的队列一致，那么就可以通过重写一个topic，调大topic中队列的数量，如原来的队列个数只有4，那么可以创建一个新的topic，设置队列的个数为8，并且原来的消费者对消息不消费，而是做一个转发功能，将4个队列的topic的数据转发到8个队列的topic中，那么在消费者组中，其个数就可以设置成8，那么这样子就很好的处理消息积压的问题了。

数据的搬运可以在具体的消费者代码里面去编写，主要功能有接收四个topic队列的数据，然后转发到八个topic的队列中，最后再写一个消费者去消费八个队列topic的消息

2.4，如何保证顺序消费

这里的顺序消息只能保证局部有序，而不是全局有序。在rocketmq内部，在生产者端，消息会根据id做一个取模运算，会将同一个区取模运算的值放入一个队列里面，在消费者端，会锁定队列消费，就是会先消费完一个队列再消费下一个队列，从而保证单个队列消费的有序性

2.5，rocketmq的持久化

rocketmq为了保证消息的安全性，在broker内部都会做一个持久化的操作，首先当生产者将消息发送到broker之后，会现将消息存储到 coimmit 文件中，每个topic都会有对应的commit文件，每个文件大小为1g，如果消息满了则会创建新的文件，文件的格式为二进制格式。

在消费者中，会有一个 comsumeQueue 文件，改文件不存数据，只存索引信息，如存一些偏移量等，在消费时可以更快的定位到commit文件中的数据，随后去消费里面的数据，并且可以通过Tag标签去过滤消息

除了上面两个文件之外，还有维护一个index文件，内部会记录Commit日志的偏移量等

2.6，死信队列

当broker和consumer之间重试16次之后，消息依旧没能被消费，那么消息就会加入到死信队列中。一个私信队列会对应一个消费者组，其perm对应的权限值为2。死信队列的消息默认不会被消费，而是需要开发者自身去处理该队列中的数据。

并且私信队列中消息的有效期也是三天，可以在broker.conf配置文件设置，当超过这个时间，消息都会被删除。

2.7，消息的幂等性

在rocketmq中，消息的幂等性为 at least once 至少被消费一次。官方建议使用里面的key去做幂等性，key是一个唯一值，就是一个唯一id。除了这些方式之外，在分布式场景下，也可以开率分布式锁这些做幂等。

3，rocketmq高性能的设计

3.1，零拷贝技术

零拷贝是操作系统层面的一种加速文件读写的操作机制，可以通过这种零拷贝的形式提升IO操作的性能。在java中，主要是通过这种 fileChannel 的方式实现零拷贝，其具体实现由 mmap和sendFile 两种形式

以一个文件的拷贝为例，正常来说，需要从用户态切换到内核态，然后再去执行io操作，并且需要通过cpu的调度，从磁盘中将文件加载到内存，再加载到网卡。而在引入零拷贝技术之后，可以让channel代替cpu去做io操作，cpu只需要给channel对应的权限即可。在操作系统层面，就是利用这种DMA技术，将原来四次的cpu拷贝，变成了两次，从而提高整体性能。

3.2，顺序写技术

本人在写过一个顺序io和随机io的文章:https://zhenghuisheng.blog.csdn.net/article/details/129080088 ，顺序写可以减少磁头的移动去寻址，不管是插入数据还是查询数据，都可以提升其性能，并且可以减少磁盘的碎片。

3.3，刷盘机制

rocketmq为了保证数据的安全性，在broker中会持久化到commitlog中，在刷盘时有两种方式，分别是：同步刷盘和异步刷盘 ，默认采用的刷盘机制时异步刷盘

flushDiskType=ASYNC_FLUSH