RocketMQ问答

心已赠人 2023-02-25 03:39 22阅读 0赞

### 文章目录 ###

*   *  [一、为什么要使用MQ？我们系统中如何使用MQ的？为什么用他？介绍一下RocketMQ。][MQ_MQ_RocketMQ]
     *  [二、RocketMQ的架构是怎么样的？][RocketMQ]
        
         *  [引申1：RocketMQ如何做到高可用？][1_RocketMQ]
         *  [引申2：RocketMQ如何保证消息不被重复消费？][2_RocketMQ]
         *  [引申3：如何保证消息的可靠传输？][3]
         *  [引申4：为什么是NameServer而不是zookeeper？][4_NameServer_zookeeper]
         *  [引申5：RocketMQ的消息消费模式和消息传输模式。][5_RocketMQ]
     *  [三、RocketMQ工作的流程是怎么样的？][RocketMQ 1]
        
         *  [引申1：RocketMQ如何实现负载均衡？][1_RocketMQ 1]
         *  [引申2：RocketMQ中的顺序消息是怎么实现的？][2_RocketMQ 1]
         *  [引申3：RocketMQ中的事务消息是怎么实现的？][3_RocketMQ]
     *  [四、RocketMQ的消息存储是怎么样的？][RocketMQ 2]
     *  [五、如何解决消息队列的延时以及过期失效问题？消息队列满了以后该怎么处理？有几百万消息持续积压几小时，说说怎么解决？][Link 1]

通过一段时间的学习，我对RocketMQ有了一定的了解，但是感觉还是有很多疑惑。目前很多知识也还是停留在理论阶段，需要更多的实践。期望能和小伙伴们一起交流交流，获得更多的收获。

## 一、为什么要使用MQ？我们系统中如何使用MQ的？为什么用他？介绍一下RocketMQ。 ##

MQ最基本的能力是：异步、削峰、解耦。  
异步是为了减少链路长度，进而减少请求响应时间。如A–>B–>C，引入MQ后变成A–>MQ,MQ–>B–>C，从A的角度看，链路变短了，响应时间也变短了。  
削峰是为了处理短时间大量请求积压，为了保障短时间大量请求的正常访问而增加很多机器比较浪费资源。可以将请求丢到MQ中，由消费者正常消费。MQ允许一定的消息积压。  
解耦是为了解除服务间的依赖，MQ消息订阅的机制，让增加一个新的后续处理变得非常简单。例如A–>B，要增加一个A–>C，有了MQ后A不需要进行修改。变成了A–>MQ，MQ–>B，MQ–>C。  
我们系统中主要用到了RocketMQ和kafka。RocketMQ应用于各个业务系统，kafka应用于大数据系统做日志采集。业务系统的一个应用示例：用户拨打电话，这个行为会写入MQ，其他系统监听这个topic，进行后续的逻辑处理。  
MQ的选型需要考虑这些因素：开发语言、单机吞吐量、时效性、可用性、特性。（如顺序消息、事务消息、是否支持积压等）  
RocketMQ基于java开发，阿里出品，现在由Apache维护，单机吞吐量十万级，高可用，功能完备。  
kafka在大数据领域广泛应用，支持度比较好。没有支持消息查询，有可能消息丢失等。我们主要用于日志采集。

## 二、RocketMQ的架构是怎么样的？ ##

RocketMQ架构上主要分为四个部分，分别是：Producer消息生产者；Consumer消息消费者；NameServer路由注册中心；Broker是MQ服务器，负责消息存储、过滤、查询、高可用等。  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JfanVzdA_size_16_color_FFFFFF_t_70][]  
RocketMQ的架构从NameServer入手会好理解一点。NameServer是一个无状态的节点，可以集群部署，节点之间不互相通信。NameServer主要做两方面的事情：Broker管理、路由管理。对应上图，其实就是NameServer–Broker，NameServer–Producer、Consumer。  
Broker管理是指，NameServer接收Broker的注册信息，并进行保存，同时提供心跳机制，监测Broker是否存活。  
路由管理是指，客户端（Producer、Consumer）通过与某一台NameServer建立长连接，就能从NameServer上获知整个Broker集群的信息。  
Producer在发送消息时，会优先读取本地缓存中的Topic、Broker等信息，将消息发送至对应的Broker。Producer启动时会开启心跳，定期从某个NameServer上获取Broker信息更新本地缓存。如果发生发送失败，会主动请求NameServer获取Broker信息。  
Consumer在消费消息时，逻辑也类似Producer。但有一点需要注意的是：Producer只会与Broker中的Master节点建立长连接，而Consumer会同时与Master和Slave建立长连接。当Consumer从Master上获取消息时，Master会建议Producer下一次从Master还是Slave获取消息。

### 引申1：RocketMQ如何做到高可用？ ###

RockerMQ的高可用体现在集群部署、主从同步、失败重试、失败剔除、定时更新。  
Producer、Consumer的集群部署使得收发两端不会成为单点，某台机器有问题不会影响集群使用。  
NameServer的集群部署使得Broker的信息得到冗余，也避免了单点。  
Broker的集群部署+主从同步使得信息能够得到备份。这里需要注意的是Broker数据写入磁盘支持同步和异步，异步场景下如果发生断电，在内存中还未写入磁盘的数据会丢失。主从同步由于是多线程异步处理的，可能存在毫秒级的延迟。  
失败重试、失败剔除、定时更新是指，消息的发送、消费、建立连接等各个环节，RocketMQ都提供了这样的机制，最大可能保证高可用。

### 引申2：RocketMQ如何保证消息不被重复消费？ ###

RockerMQ为了保证Broker的高效，没有在Broker层面上保证消息不被重复消费。消息的幂等消费应当由消费者来保证。消费者通过业务字段校验等方式保证消息的幂等。

### 引申3：如何保证消息的可靠传输？ ###

可靠传输是指，在发送端、消息队列、消费端都不会出现数据丢失。RocketMQ提供了一些策略进行保证。  
发送端，提供了事务消息的机制。  
消息队列端，提供了主从同步机制、同步刷盘机制。  
消费端，消息队列端能不丢消息，消费端这里只要提供提供消息重试、offset重置等能力就能保证消息一定被消费到。

### 引申4：为什么是NameServer而不是zookeeper？ ###

NameServer相比zookeeper更加简洁高效，当然功能也更少。zookeeper提供了选主的功能，然而RocketMQ的主从是通过命名确定的，相同命名的broker组成主从，id为0的是主。  
早期版本的RocketMQ就是用的zookeeper，但是NameServer更加符合RocketMQ的架构。

### 引申5：RocketMQ的消息消费模式和消息传输模式。 ###

Broker与Consumer之间的消息传送有两种方式：推模式、拉模式  
推模式：Broker向Consumer推送消息  
拉模式：Consumer主动向Broker拉消息  
RocketMQ的推模式是基于拉模式，在拉模式上包装了一层，一个拉取任务完成后开始下一个拉取任务。  
一个消费者组可以包含多个消费者，每个消费者都可以订阅多个主题。  
消费者组的消费模式有：集群模式、广播模式  
集群模式：topic下的同一条消息只允许被同一个消费者组下的一个消费者消费。  
广播模式：topic下的同一条消息可以被同一个消费者组下的所有消费者消费。

## 三、RocketMQ工作的流程是怎么样的？ ##

1.启动NameServer，NameServer起来后监听端口，等待Broker、Producer、Consumer连上来，相当于一个路由控制中心。  
2.Broker启动，跟所有的NameServer保持长连接，定时发送心跳包。心跳包中包含当前Broker信息(IP+端口等)以及存储所有Topic信息。注册成功后，NameServer集群中就有Topic跟Broker的映射关系。  
3.收发消息前，先创建Topic，创建Topic时需要指定该Topic要存储在哪些Broker上，也可以在发送消息时自动创建Topic。  
4.Producer发送消息，启动时先跟NameServer集群中的其中一台建立长连接，并从NameServer中获取当前发送的Topic存在哪些Broker上，轮询从队列列表中选择一个队列，然后与队列所在的Broker建立长连接从而向Broker发消息。  
5.Consumer跟Producer类似，跟其中一台NameServer建立长连接，获取当前订阅Topic存在哪些Broker上，然后直接跟Broker建立连接通道，开始消费消息。

### 引申1：RocketMQ如何实现负载均衡？ ###

Producer通过轮询本地缓存的queue数组的方式来做负载均衡。有普通模式和Broker故障延迟机制可以选择，Broker故障延迟机制：如果某Broker发生故障，一段时间内都不会尝试使用该Broker，如果该Broker已恢复，则从剔除列表中将此Broker移除。  
Consumer的负载均衡相对复杂，因为Consumer实际做的是 消息拉取 + 消息消费处理，这里的“+”号在RocketMQ中是通过回调函数、异步来实现的，实现了消息拉取和消息消费处理的解耦。  
消息拉取时会先获取拉取目标信息，在这里做了负载均衡。

### 引申2：RocketMQ中的顺序消息是怎么实现的？ ###

顺序消息实际上是改造普通消息的负载均衡策略，由轮询改为指定。  
发送时，只需要改变顺序消息中【负载均衡，获取Queue信息】这一步，改为【通过指定的select方法获取queue信息】即可保证消息按照我们指定的规则存储到对应的queue上，便于消费时按序消费。  
消费时，由于消息拉取和消息消费是异步的，需要分别加锁才能实现顺序消费。

### 引申3：RocketMQ中的事务消息是怎么实现的？ ###

RocketMQ采用了2PC的思想来实现了提交事务消息，同时增加一个补偿逻辑来处理二阶段超时或者失败的消息。  
![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JfanVzdA_size_16_color_FFFFFF_t_70 1][]  
1.发送消息（half消息）。  
2.服务端响应消息写入结果。  
3.根据发送结果执行本地事务（如果写入失败，此时half消息对业务不可见，本地逻辑不执行）。  
4.根据本地事务状态执行Commit或者Rollback（Commit操作生成消息索引，消息对消费者可见）  
5.对没有Commit/Rollback的事务消息（pending状态的消息），从服务端发起一次“回查”  
6.Producer收到回查消息，检查回查消息对应的本地事务的状态。  
7.根据本地事务状态，重新Commit或者Rollback。  
RocketMQ内部采用替换主题的方式来实现half消息对用户的不可见。

## 四、RocketMQ的消息存储是怎么样的？ ##

RocketMQ的消息存储在文件中，主要涉及CommitLog消息主体文件、ConsumeQueue消息消费队列信息、IndexFile索引文件。  
Producer端写入的消息主体内容,消息内容不是定长的。  
ConsumeQueue消息消费队列，引入的目的主要是提高消息消费的性能。每个Topic包含多个Queue，每个Queue有一个消息文件。由于不同Topic的消息都存储在CommitLog中，如果要遍历commitlog文件中根据topic检索消息是非常低效的。Consumer即可根据ConsumeQueue来查找待消费的消息。  
IndexFile索引文件是为了加速消息的检索性能，提供了根据key或时间区间来查询消息的方法。  
RocketMQ主要通过MappedByteBuffer对文件进行读写操作，利用了NIO中的FileChannel模型将磁盘上的物理文件直接映射到用户态的内存地址中，将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率。  
（疑惑：commitlog默认大小为1G，RocketMQ会对所有文件创建内存映射，当消息大量积压时，是不是对机器的内存消耗特别大？）  
同步刷盘：只有在消息真正持久化至磁盘后RocketMQ的Broker端才会真正返回给Producer端一个成功的ACK响应。同步刷盘对MQ消息可靠性来说是一种不错的保障，但是性能上会有较大影响，一般适用于金融业务应用该模式较多。  
异步刷盘：只要消息写入PageCache即可将成功的ACK返回给Producer端。消息刷盘采用后台异步线程提交的方式进行，降低了读写延迟，提高了MQ的性能和吞吐量。  
RocketMQ不会永久存储文件，会有文件过期机制、容量保护机制删除文件。

## 五、如何解决消息队列的延时以及过期失效问题？消息队列满了以后该怎么处理？有几百万消息持续积压几小时，说说怎么解决？ ##

这个问题在网上有一些解答，但感觉答案有点令人疑惑。疑惑点是操作太多，线上这样做风险比较大。参考这两篇文章  
https://www.cnblogs.com/yuxiang1/p/10579633.html  
https://mp.weixin.qq.com/s/B0STtECjefDDZwUBFeSqEg  
消费者处理慢甚至不消费，可能的原因有这些：  
1.消费者出现bug，甚至挂了。  
2.对消息量级评估有误，消费速度远小于生产速度。  
3.消费者逻辑复杂，响应慢。

1.对于bug，可以紧急修复的修复后上线。如果不能紧急修复，为了避免影响MQ集群中其他Topic消息的消费，甚至可以直接ack， 后面慢慢修复。等修复好后调整offset重新消费，重新消费的幂等由消费者保证。  
2.对于量级评估失误，可以进行消费者扩容。消费者比较好的一种做法是尽量的轻量级，只做MQ的消费，业务逻辑的处理交由其他服务完成。这样消费者对资源消耗比较小，比较好扩容。  
3.逻辑复杂、响应慢的问题可以用异步处理的方式来解。消费者收到消息进行落库，然后直接ack，逻辑处理交由线程池异步处理。这样对MQ比较友好，基本不会有积压。（疑惑：RocketMQ单机吞吐量10W+，远超数据库。引入MQ本身也有削峰功能，这里反而用数据库来保障MQ的无积压，舍弃了MQ的失败重试机制，需要自己实现失败重试。是好设计吗？有更好方案吗？）  
4.可以通过批量方式消费增加消费者处理速度。可以通过设置consumer的参数来控制每次消费消息的条数。

[MQ_MQ_RocketMQ]: https://www.pianshen.com/article/38821165746/#MQMQRocketMQ_3
[RocketMQ]: https://www.pianshen.com/article/38821165746/#RocketMQ_14
[1_RocketMQ]: https://www.pianshen.com/article/38821165746/#1RocketMQ_24
[2_RocketMQ]: https://www.pianshen.com/article/38821165746/#2RocketMQ_31
[3]: https://www.pianshen.com/article/38821165746/#3_34
[4_NameServer_zookeeper]: https://www.pianshen.com/article/38821165746/#4NameServerzookeeper_40
[5_RocketMQ]: https://www.pianshen.com/article/38821165746/#5RocketMQ_44
[RocketMQ 1]: https://www.pianshen.com/article/38821165746/#RocketMQ_55
[1_RocketMQ 1]: https://www.pianshen.com/article/38821165746/#1RocketMQ_62
[2_RocketMQ 1]: https://www.pianshen.com/article/38821165746/#2RocketMQ_67
[3_RocketMQ]: https://www.pianshen.com/article/38821165746/#3RocketMQ_72
[RocketMQ 2]: https://www.pianshen.com/article/38821165746/#RocketMQ_85
[Link 1]: https://www.pianshen.com/article/38821165746/#_97
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JfanVzdA_size_16_color_FFFFFF_t_70]: /images/20230209/019cde0517fa497b94273884efb7cac8.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2JfanVzdA_size_16_color_FFFFFF_t_70 1]: /images/20230209/d98d7528655d4465bab6979ff30e2805.png