必知必会：RocketMQ

Posted 2023-06-18 时光编辑师

• 1.优点：

应用解耦、异步处理、流量削峰。

• 2.消息模型：

发布-订阅模型。
一个消息可以被多个消费组消费；
一个消费组中包含多个消费者，同一个消费组的每个消费者负责一部分消息。

• 3.消息模式：

有两种，集群消息和广播消息，默认是集群消息。

集群消息：一个消费组共同消费一个主题的多个队列，每个队列只能被一个消费者消费。

广播消息：消息会发给消费组的每一个消费者消费。

• 4.基本架构：

NameServer、Broker、Producer、Consumer，为了保证高可用，这几个部分一般都是集群部署的。

NameServer：是一个无状态的服务器，每个NameServer相互独立，它们之间不存在任何信息交互。
主要功能是 1.和Broker结点保持长连接；2.维护Topic的路由信息。

Broker：负责存储和转发消息。
Borker内部维护着很多ConsumerQueue，用来存储消息的索引，真正消息存储在CommitLog日志文件中。
每个Borker与所有的NameServer保持长连接和心跳，并定时将Topic信息同步到NameServer。

Producer：消息生产者，消息由Producer通过负载均衡模式发送到Broker集群。
支持三种发送消息的方式：同步、异步、单项发送。

Consumer：消息消费者，支持两种消费模式：Pull和Push，支持集群消息和广播消息。
Pull：主要从消息服务器拉取消息，主要批量拉取到消息，用户就会启动消费过程。
Push：Push要先注册消费监听者，当监听器触发后就会开始消费消息。推送型消费者封装了消息的拉取、消息进度和其他内部维护工作。

5.消息在哪些阶段可能会丢失？

要从消息生产、消息存储、消息消费阶段考虑。

消息生产阶段：通过请求确认机制，来保证消息的可靠性传递。

消息存储阶段：1.消息持久化到CommitLog日志文件中即使宕机也可以恢复后再消费。刷盘机制分为同步刷盘和异步刷盘。
2.Broker主从模式来保证高可用，Broker支持同步复制和异步复制。
生产者的消息都是发送到Master。Slave可以同步复制Master或异步复制Master。
消费者可以从Master读取消息，也可以从Slave读取消息。
同步复制可保证即使Master宕机，Slave也有消息备份，保证消息不会丢失。

消息消费阶段：消费者需在收到消息并执行完所有消息业务逻辑后，再发送消息确认。
这样可以保证如果消息处理失败了，没有确认，再去从消息队列拉取的消息就还会是原来那条。

6.消息重复怎么处理？

业务幂等、消息去重。

业务幂等：即执行一次和多次结果一样。

消息去重：用消息的唯一编号来保证。

• 7.消息积压怎么处理？

需要提高消费能力，如消费者扩容，消息迁移Queue扩容。

消费者扩容：如果当前Topic的消息队列的数量大于消费者数据，可以增加消费者来提供消费能力。
消息迁移Queue扩容：新建一个临时的Topic，临时的Topic多设置一些队列，将当前Topic的消息转发到临时的Topic中，然后再用扩容的消费者去消费新的Topic，后续再恢复。

• 8.顺序消息怎么实现？

顺序消息就是消息的消费顺序要和产生顺序相同。顺序消息又分为全局顺序消息和局部顺序消息。

全局顺序消息：消除并发，读写队列设置为1，这样就牺牲了RocketMQ高并发、高吞吐的特性。

局部顺序消息：
（1）保证生产者把同一个ID的消息发往一个消息队列，发送端使用消息选择器类MessageQueueSelector控制把消息发往哪一个消息队列。
（2）由于每个队列的消息都是给同一个消费者消费的，这里只需保证消费者不要并发处理，可使用MessageListenerOrderly类处理。

使用消息内存队列：对关键字hash取模，保证同一ID的不同状态的数据能到一个内存队列。

• 9.Broker是怎么保存数据的？

主要的存储文件有CommitLog文件、ConsumerQueue文件、IndexFile文件。

CommitLog：存储Producer端写入的消息主体内容。每个文件默认是1G，当文件满了就写入下一个文件，文件名记录了固定20长度的，记录了起始偏移量。

ConsumerQueue：可以看成是基于Topic的CommitLog索引文件；
保存有某个Topic下队列消息在CommitLog中的起始物理偏移量offset，消息大小size和消息tag的HashCode值。

IndexFile：索引文件，提供了一个可以通过key或时间区间来查询消息的方法。