一、MQ概述

介绍：

用途：

               MQ可以将系统的超量请求暂存其中，以便系统后期可以慢慢进行处理，从而避免了请求的丢失或系统被压垮。

             上游系统对下游系统的调用若为同步调用，则会大大降低系统的吞吐量与并发度，且系统耦合度太高。 而异步调用则会解决这些问题。所以两层之间若要实现由同步到异步的转化，一般性做法就是，在这两层间添加一个MQ层。

MQ产品：

        链接

二、RocketMQ概述

三、RocketMQ组成与架构

（1）组成

1 消息（Message）

 2 主题（Topic）

3 标签（Tag）

Topic ：货物

tag= 上海

tag= 江苏 tag= 浙江

------- 消费者 -----

topic= 货物 tag = 上海

topic= 货物 tag = 上海 | 浙江

topic= 货物 tag = *

4 队列（Queue）

5分片（Sharding）

6 消息标识（MessageId/Key）

• msgId：由producer端生成，其生成规则为： producerIp + 进程pid + MessageClientIDSetter类的ClassLoader的hashCode + 当前时间 + AutomicInteger自增计数器
• offsetMsgId：由broker端生成，其生成规则为：brokerIp + 物理分区的offset（Queue中的偏移量）
• key：由用户指定的业务相关的唯一标识

（2）架构

1 Producer

数据写入到MQ中的过程

2 Consumer

从MQ中获取到数据的过程

         消费者组中Consumer的数量应该小于等于订阅Topic的Queue数量。如果超出Queue数量，则多出的Consumer将不能消费消息。

 不过，一个Topic类型的消息可以被多个消费者组同时消费。

注意，

1 ）消费者组只能消费一个 Topic 的消息，不能同时消费多个 Topic 消息

2 ）一个消费者组中的消费者必须订阅完全相同的 Topic

3 Name Server

注意，这是与其它像 zk 、 Eureka 、 Nacos 等注册中心不同的地方。

这种 NameServer 的无状态方式，有什么优缺点：

优点： NameServer 集群搭建简单，扩容简单。

缺点：对于 Broker ，必须明确指出所有 NameServer 地址。否则未指出的将不会去注册。也正因为如此，NameServer 并不能随便扩容。因为，若 Broker 不重新配置，新增的 NameServer 对于 Broker 来说是不可见的，其不会向这个 NameServer 进行注册。

扩展：

        对于RocketMQ 日常运维工作，例如 Broker 升级，需要停掉 Broker 的工作。 OP 需要怎么 做？

        OP需要将 Broker 的读写权限禁掉。一旦 client(Consumer 或 Producer) 向 broker 发送请求，都会收到broker 的 NO_PERMISSION 响应，然后 client 会进行对其它 Broker 的重试。

        当OP 观察到这个 Broker 没有流量后，再关闭它，实现 Broker 从 NameServer 的移除。

OP ：运维工程师

SRE ： Site Reliability Engineer ，现场可靠性工程师

扩展：

        1） Push 模型：推送模型。其实时性较好，是一个 “ 发布 - 订阅 ” 模型，需要维护一个长连接。而长连接的维护是需要资源成本的。该模型适合于的场景：

• 实时性要求较高
• Client数量不多，Server数据变化较频繁

        2） Pull 模型：拉取模型。存在的问题是，实时性较差。

        3） Long Polling 模型：长轮询模型。其是对 Push 与 Pull 模型的整合，充分利用了这两种模型的优势，屏蔽了它们的劣势。

扩展： Zookeeper Client 是如何选择 Zookeeper Server 的？

        简单来说就是， 经过两次Shuffle，然后选择第一台Zookeeper Server。 详细说就是，将配置文件中的zk server 地址进行第一次shuffle，然后随机选择一个。这个选择出的一般都是一个hostname。然后获取到该 hostname 对应的所有 ip ，再对这些 ip 进行第二次 shuffle ，从 shuffle 过的结果中取第一个 server 地址进行连接。

4 Broker

Kafka 0.8 版本之后， offset 是存放在 Broker 中的，之前版本是存放在 Zookeeper 中的。

5.集群部署：

工作过程：

           1）启动NameServer，NameServer启动后开始监听端口，等待Broker、Producer、Consumer连接。

二、Topic的创建模式

• 集群模式：该模式下创建的Topic在该集群中，所有Broker中的Queue数量是相同的。
• Broker模式：该模式下创建的Topic在该集群中，每个Broker中的Queue数量可以不同。

三、读/写队列

（3）集群搭建理论

一、数据复制与刷盘策略

复制策略:

        复制策略是Broker的Master与Slave间的数据同步方式。分为同步复制与异步复制：

         同步复制：消息写入master后，master会等待slave同步数据成功后才向producer返回成功ACK

        异步复制：消息写入master后，master立即向producer返回成功ACK，无需等待slave同步数据成 功

异步复制策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量

刷盘策略:

        刷盘策略指的是broker中消息的落盘方式，即消息发送到broker内存后消息持久化到磁盘的方式。分为 同步刷盘与异步刷盘：

同步刷盘：当消息持久化到broker的磁盘后才算是消息写入成功。

异步刷盘：当消息写入到broker的内存后即表示消息写入成功，无需等待消息持久化到磁盘。

1）异步刷盘策略会降低系统的写入延迟，RT变小，提高了系统的吞吐量 2）消息写入到Broker的内存，一般是写入到了PageCache

3）对于异步 刷盘策略，消息会写入到PageCache后立即返回成功ACK。但并不会立即做落盘操 作，而是当PageCache到达一定量时会自动进行落盘。

if (FlushDiskType.SYNC_FLUSH == defaultMessageStore.getMessageStoreConfig().getFlushDiskType()) {
            this.flushCommitLogService = new GroupCommitService();  //同步
        } else {
            this.flushCommitLogService = new FlushRealTimeService(); // 异步
        }

        this.commitLogService = new CommitRealTimeService();   异步转存

二、Broker集群模式

多Master ：

前提:

        Master 都配置了 RAID 磁盘阵列。如果没有配置，一旦出现某 Master 宕机，则会发生大量消息丢失的情况。

多Master多Slave模式-异步复制:

        异步复制:

                消息写入master成功后，master立即向 producer返回成功ACK，无需等待slave同步数据成功。

        该模式的最大特点之一是，当master宕机后slave能够自动切换为master。不过由于slave从master的同步具有短暂的延迟（毫秒级），所以当master宕机后，这种异步复制方式可能会存在少量消息的丢失问题。

Slave 从 Master 同步的延迟越短，其可能丢失的消息就越少

对于 Master 的 RAID 磁盘阵列，若使用的也是异步复制策略，同样也存在延迟问题，同样也可能会丢失消息。但RAID 阵列的秘诀是微秒级的（因为是由硬盘支持的），所以其丢失的数据量会更少。

多Master多Slave模式-同步双写 :

        该模式是多Master多Slave模式的同步复制实现。所谓同步双写，指的是消息写入master成功后， master会等待slave同步数据成功后才向producer返回成功ACK，即master与slave都要写入成功后才会返回成功ACK，也即双写。

        该模式与异步复制模式相比，优点是消息的安全性更高，不存在消息丢失的情况。但单个消息的RT略高，从而导致性能要略低（大约低10%）。

        优点：数据与服务都无单点故障，Master宕机情况下，消息无延迟，服务可用性与数据可用性都非常高；
        缺点：性能比异步复制模式略低（大约低10%左右），发送单个消息的RT会略高，且目前版本在主节点宕机后，备机不能自动切换为主机。

        一般会为Master配置RAID10磁盘阵列，然后再为其配置一个Slave。即利用了RAID10磁盘阵列的高效、安全性，又解决了可能会影响订阅的问题。

1 ） RAID 磁盘阵列的效率要高于 Master-Slave 集群。因为 RAID 是硬件支持的。也正因为如此， 所以RAID 阵列的搭建成本较高。

2 ）多 Master+RAID 阵列，与多 Master 多 Slave 集群的区别是什么？

        多Master+RAID 阵列，其仅仅可以保证数据不丢失，即不影响消息写入，但其可能会影响到 消息的订阅。但其执行效率要远高于 多Master多Slave集群

        多Master 多 Slave 集群，其不仅可以保证数据不丢失，也不会影响消息写入。其运行效率要低于 多Master+RAID阵列

master/slave 高保证数据不丢失，高消费可用，多master是高可用。

RAID技术详解与总结

四、RocketMQ安装

五、RocketMQ工作原理

一、消息的生产

1. Producer发送消息之前，会先向NameServer发出获取消息Topic的路由信息的请求
2. NameServer返回该Topic的路由表及Broker列表
3. Producer根据代码中指定的Queue选择策略，从Queue列表中选出一个队列，用于后续存储消息 Produer对消息做一些特殊处理，例如，消息本身超过4M，则会对其进行压缩
4. Producer向选择出的Queue所在的Broker发出RPC请求，将消息发送到选择出的Queue

路由表：实际是一个Map， key为Topic名称，value是一个QueueData实例列表。QueueData并不 是一个Queue对应一个QueueData，而是一个Broker中该Topic的所有Queue对应一个 QueueData。即，只要涉及到该Topic的Broker，一个Broker对应一个QueueData。QueueData中 包含brokerName。简单来说，路由表的key为Topic名称，value则为所有涉及该Topic的 BrokerName列表。

Broker列表：其实际也是一个Map。 key为brokerName，value为BrokerData。一个Broker对应一 个BrokerData实例，对吗？不对。一套brokerName名称相同的Master-Slave小集群对应一个 BrokerData。BrokerData中包含brokerName及一个map。该map的key为brokerId，value为该 broker对应的地址。 brokerId为0表示该broker为Master，非0表示Slave。

该算法存在一个问题：由于某些原因，在某些Broker上的Queue可能 投递延迟较严重。从而导致 Producer的缓存队列中出现较大的 消息积压，影响消息的投递性能。

该算法也存在一个问题：消息在Queue上的分配不均匀。投递延迟小的Queue其可能会存在大量的消息。而对该Queue的消费者压力会增大，降低消息的消费能力，可能会导致MQ中消息的堆积。

二、消息的存储

1. abort：该文件在Broker启动后会自动创建，正常关闭Broker，该文件会自动消失。若在没有启动 Broker的情况下，发现这个文件是存在的，则说明之前Broker的关闭是非正常关闭。
2. checkpoint：其中存储着commitlog、consumequeue、index文件的最后刷盘时间戳
3. commitlog：其中存放着commitlog文件，而消息是写在commitlog文件中的
4. config：存放着Broker运行期间的一些配置数据
5. consumequeue：其中存放着consumequeue文件，队列就存放在这个目录中
6. index：其中存放着消息索引文件indexFile
7. lock：运行期间使用到的全局资源锁

public DefaultMessageStore(final MessageStoreConfig messageStoreConfig, final BrokerStatsManager brokerStatsManager,
        final MessageArrivingListener messageArrivingListener, final BrokerConfig brokerConfig) throws IOException {
        // 消息到达监听  在消费长轮询中使用
        this.messageArrivingListener = messageArrivingListener;
        this.brokerConfig = brokerConfig;
        this.messageStoreConfig = messageStoreConfig;
        // broker状态管理
        this.brokerStatsManager = brokerStatsManager;
        // 内存映射文件创建服务
        this.allocateMappedFileService = new AllocateMappedFileService(this);
        this.commitLog = new CommitLog(this);
        // topic 与 多个comsumerqueue对应关系
        this.consumeQueueTable = new ConcurrentHashMap<>(32);
        // 刷盘服务
        this.flushConsumeQueueService = new FlushConsumeQueueService();
        // 清理commit log
        this.cleanCommitLogService = new CleanCommitLogService();
        // 清理ConsumeQueue
        this.cleanConsumeQueueService = new CleanConsumeQueueService();
        //存储状态管理
        this.storeStatsService = new StoreStatsService();
        // 索引服务
        this.indexService = new IndexService(this);
        // 主从同步服务
        this.haService = new HAService(this);
        // reputMessageService是用来把commitlog中的数据写到consumerqueue和index中
        this.reputMessageService = new ReputMessageService();

        this.scheduleMessageService = new ScheduleMessageService(this);

        // 直接内存池，避免频繁创建，在异步刷盘的时候作为与pagecache的数据交换区，通过commit操作完成数据传输到pagecache
        this.transientStorePool = new TransientStorePool(messageStoreConfig);

        if (messageStoreConfig.isTransientStorePoolEnable()) {
            this.transientStorePool.init();
        }

        // 启动mappedFile创建服务
        this.allocateMappedFileService.start();
        // 启动索引服务
        this.indexService.start();

        // 在上面的reputMessageService中的reput方法中，调用下面的2个的dodispatch方法
        this.dispatcherList = new LinkedList<>();
        this.dispatcherList.addLast(new CommitLogDispatcherBuildConsumeQueue());
        this.dispatcherList.addLast(new CommitLogDispatcherBuildIndex());
    }

1.commitlog文件

说明：在很多资料中commitlog目录中的文件简单就称为commitlog文件。但在源码中，该文件 被命名为mappedFile。

        第一个文件名一定是20位0构成的。因为第一个文件的第一条消息的偏移量commitlog offset为0

        当第一个文件放满时，则会自动生成第二个文件继续存放消息。假设第一个文件大小是 1073741820字节（1G = 1073741824字节），则第二个文件名就是00000000001073741824。

        以此类推，第n个文件名应该是前n-1个文件大小之和。

        一个Broker中所有mappedFile文件的commitlog offset是连续的

mappedFile文件是顺序读写的文件，所有其访问效率很高

无论是SSD磁盘还是SATA磁盘，通常情况下，顺序存取效率都会高于随机存取

        需要注意到，消息单元中是包含Queue相关属性的。所以，我们在后续的学习中，就需要十分 留意commitlog与queue间的关系是什么？

        一个mappedFile文件中第m+1个消息单元的commitlog，offset偏移量 L(m+1) = L(m) + MsgLen(m) (m >= 0)

2.consumequeue

目录与文件:

        为了提高效率，会为每个Topic在~/store/consumequeue中创建一个目录，目录名为Topic名称。在该 Topic目录下，会再为每个该Topic的Queue建立一个目录，目录名为queueId。每个目录中存放着若干 consumequeue文件，consumequeue文件是commitlog的索引文件，可以根据consumequeue定位到具 体的消息。

        consumequeue文件名也由20位数字构成，表示当前文件的第一个索引条目的起始位移偏移量。与 mappedFile文件名不同的是，其后续文件名是固定的。因为consumequeue文件大小是固定不变的。

索引条目:

        每个consumequeue文件可以包含30w个索引条目，每个索引条目包含了三个消息重要属性：消息在 mappedFile文件中的偏移量CommitLog Offset、消息长度、消息Tag的hashcode值。这三个属性占20 个字节，所以每个文件的大小是固定的30w * 20字节。

 一个consumequeue文件中所有消息的Topic一定是相同的。但每条消息的Tag可能是不同的。

3 对文件的读写

消息写入:

一条消息进入到Broker后经历了以下几个过程才最终被持久化。

• Broker根据queueId，获取到该消息对应索引条目要在consumequeue目录中的写入偏移量，即 QueueOffset
• 将queueId、queueOffset等数据，与消息一起封装为消息单元
• 将消息单元写入到commitlog
• 同时，形成消息索引条目
• 将消息索引条目分发到相应的consumequeue

消息拉取:

当Consumer来拉取消息时会经历以下几个步骤：

        1.Consumer获取到其要消费消息所在Queue的消费偏移量offset，计算出其要消费消息的 消息offset

        消费offset即消费进度，consumer对某个Queue的消费offset，即消费到了该Queue的第几 条消息 消息offset = 消费offset + 1

        2.Consumer向Broker发送拉取请求，其中会包含其要拉取消息的Queue、消息offset及消息 Tag。

        3.Broker计算在该consumequeue中的queueOffset。

queueOffset = 消息offset * 20字节

        4.从该queueOffset处开始向后查找第一个指定Tag的索引条目。

        5.解析该索引条目的前8个字节，即可定位到该消息在commitlog中的commitlog offset

        6.从对应commitlog offset中读取消息单元，并发送给Consumer

性能提升:

        RocketMQ中，无论是消息本身还是消息索引，都是存储在磁盘上的。其不会影响消息的消费吗？当然 不会。其实RocketMQ的性能在目前的MQ产品中性能是非常高的。因为系统通过一系列相关机制大大 提升了性能。

        首先，RocketMQ对文件的读写操作是通过mmap零拷贝进行的，将对文件的操作转化为直接对内存地址进行操作，从而极大地提高了文件的读写效率。

        其次，consumequeue中的数据是顺序存放的，还引入了PageCache的预读取机制，使得对 consumequeue文件的读取几乎接近于内存读取，即使在有消息堆积情况下也不会影响性能。

PageCache机制，页缓存机制，是OS对文件的缓存机制，用于加速对文件的读写操作。一般来说，程序对文件进行顺序读写的速度几乎接近于内存读写速度，主要原因是由于OS使用 PageCache机制对读写访问操作进行性能优化，将一部分的内存用作PageCache。

写操作：OS会先将数据写入到PageCache中，随后会以异步方式由pdflush（page dirty flush) 内核线程将Cache中的数据刷盘到物理磁盘

读操作：若用户要读取数据，其首先会从PageCache中读取，若没有命中，则OS在从物理磁 盘上加载该数据到PageCache的同时，也会顺序对其相邻数据块中的数据进行预读取。

三、indexFile

        除了通过通常的指定Topic进行消息消费外，RocketMQ还提供了根据key进行消息查询的功能。该查询 是通过store目录中的index子目录中的indexFile进行索引实现的快速查询。当然，这个indexFile中的索 引数据是在包含了key的消息被发送到Broker时写入的。如果消息中没有包含key，则不会写入。

1.索引条目结构

        每个Broker中会包含一组indexFile，每个indexFile都是以一个时间戳命名的（这个indexFile被创建时的时间戳）。每个indexFile文件由三部分构成：indexHeader，slots槽位，indexes索引数据。每个 indexFile文件中包含500w个slot槽。而每个slot槽又可能会挂载很多的index索引单元。

indexHeader固定40个字节，其中存放着如下数据：

​​​​​​​

• beginTimestamp：该indexFile中第一条消息的存储时间
• endTimestamp：该indexFile中最后一条消息存储时间
• beginPhyoffset：该indexFile中第一条消息在commitlog中的偏移量commitlog offse
• endPhyoffset：该indexFile中最后一条消息在commitlog中的偏移量commitlog offset
• hashSlotCount：已经填充有index的slot数量（并不是每个slot槽下都挂载有index索引单元，这 里统计的是所有挂载了index索引单元的slot槽的数量）
• indexCount：该indexFile中包含的索引单元个数（统计出当前indexFile中所有slot槽下挂载的所 有index索引单元的数量之和）

indexFile中最复杂的是Slots与Indexes间的关系。在实际存储时，Indexes是在Slots后面的，但为了便 于理解，将它们的关系展示为如下形式：

        key的hash值 % 500w的结果即为slot槽位，然后将该slot值修改为该index索引单元的indexNo，根 据这个indexNo可以计算出该index单元在indexFile中的位置。不过，该取模结果的重复率是很高的， 为了解决该问题，在每个index索引单元中增加了preIndexNo，用于指定该slot中当前index索引单元的 前一个index索引单元。而slot中始终存放的是其下最新的index索引单元的indexNo，这样的话，只要 找到了slot就可以找到其最新的index索引单元，而通过这个index索引单元就可以找到其之前的所有 index索引单元。

indexNo是一个在indexFile中的流水号，从0开始依次递增。即在一个indexFile中所有indexNo是 以此递增的。indexNo在index索引单元中是没有体现的，其是通过indexes中依次数出来的。

index索引单元默写20个字节，其中存放着以下四个属性：

• keyHash：消息中指定的业务key的hash值
• phyOffset：当前key对应的消息在commitlog中的偏移量commitlog offset
• timeDiff：当前key对应消息的存储时间与当前indexFile创建时间的时间差
• preIndexNo：当前slot下当前index索引单元的前一个index索引单元的indexNo

2 indexFile的创建

indexFile的文件名为当前文件被创建时的时间戳。这个时间戳有什么用处呢？

        根据业务key进行查询时，查询条件除了key之外，还需要指定一个要查询的时间戳，表示要查询不大于该时间戳的最新的消息，即查询指定时间戳之前存储的最新消息。这个时间戳文件名可以简化查询，提 高查询效率。具体后面会详细讲解。

indexFile文件是何时创建的？其创建的条件（时机）有两个：

• 当第一条带key的消息发送来后，系统发现没有indexFile，此时会创建第一个indexFile文件
• 当一个indexFile中挂载的index索引单元数量超出2000w个时，会创建新的indexFile。当带key的 消息发送到来后，系统会找到最新的indexFile，并从其indexHeader的最后4字节中读取到 indexCount。若indexCount >= 2000w时，会创建新的indexFile。

由于可以推算出，一个indexFile的最大大小是：(40 + 500w * 4 + 2000w * 20)字节

3 查询流程

        当消费者通过业务key来查询相应的消息时，其需要经过一个相对较复杂的查询流程。不过，在分析查询流程之前，首先要清楚几个定位计算式子：

计算指定消息key的slot槽位序号： slot槽位序号 = key的hash % 500w

计算槽位序号为n的slot在indexFile中的起始位置： slot(n)位置 = 40 + (n - 1) * 4

计算indexNo为m的index在indexFile中的位置： index(m)位置 = 40 + 500w * 4 + (m - 1) * 20

40为indexFile中indexHeader的字节数

500w * 4 是所有slots所占的字节数

流程:

四、消息的消费

        消费者从Broker中获取消息的方式有两种：pull拉取方式和push推动方式。消费者组对于消息消费的模 式又分为两种：集群消费Clustering和广播消费Broadcasting。

1.获取消费类型

拉取式消费:

        Consumer主动从Broker中拉取消息，主动权由Consumer控制。一旦获取了批量消息，就会启动消费过程。不过，该方式的实时性较弱，即Broker中有了新的消息时消费者并不能及时发现并消费。

由于拉取时间间隔是由用户指定的，所以在设置该间隔时需要注意平稳：间隔太短，空请求比 例会增加；间隔太长，消息的实时性太差

推送式消费:

        该模式下Broker收到数据后会主动推送给Consumer。该获取方式一般实时性较高。

        该获取方式是典型的发布-订阅模式，即Consumer向其关联的Queue注册了监听器，一旦发现有新的 消息到来就会触发回调的执行，回调方法是Consumer去Queue中拉取消息。而这些都是基于Consumer 与Broker间的长连接的。长连接的维护是需要消耗系统资源的。

对比:

        pull：需要应用去实现对关联Queue的遍历，实时性差；但便于应用控制消息的拉取

        push：封装了对关联Queue的遍历，实时性强，但会占用较多的系统资源

2 消费模式

广播消费:

        广播消费模式下，相同Consumer Group的每个Consumer实例都接收同一个Topic的全量消息。即每条 消息都会被发送到Consumer Group中的每个Consumer。

集群消费:

        集群消费模式下，相同Consumer Group的每个Consumer实例平均分摊同一个Topic的消息。即每条消息只会被发送到Consumer Group中的某个Consumer。

消息进度保存:

1. 广播模式：消费进度保存在consumer端。因为广播模式下consumer group中每个consumer都会 消费所有消息，但它们的消费进度是不同。所以consumer各自保存各自的消费进度。
2. 集群模式：消费进度保存在broker中。consumer group中的所有consumer共同消费同一个Topic 中的消息，同一条消息只会被消费一次。消费进度会参与到了消费的负载均衡中，故消费进度是 需要共享的。下图是broker中存放的各个Topic的各个Queue的消费进度。

3 Rebalance机制

Rebalance机制讨论的前提是：集群消费。

什么是Rebalance?

        Rebalance即再均衡，指的是，将⼀个Topic下的多个Queue在同⼀个Consumer Group中的多个 Consumer间进行重新分配的过程。

        Rebalance机制的本意是为了提升消息的并行消费能力。例如，⼀个Topic下5个队列，在只有1个消费者的情况下，这个消费者将负责消费这5个队列的消息。如果此时我们增加⼀个消费者，那么就可以给其中⼀个消费者分配2个队列，给另⼀个分配3个队列，从而提升消息的并行消费能力。

Rebalance限制：

         由于⼀个队列最多分配给⼀个消费者，因此当某个消费者组下的消费者实例数量大于队列的数量时， 多余的消费者实例将分配不到任何队列。

Rebalance危害：

        消费暂停：在只有一个Consumer时，其负责消费所有队列；在新增了一个Consumer后会触发 Rebalance的发生。此时原Consumer就需要暂停部分队列的消费，等到这些队列分配给新的Consumer 后，这些暂停消费的队列才能继续被消费。

        消费重复：Consumer 在消费新分配给自己的队列时，必须接着之前Consumer 提交的消费进度的offset 继续消费。然而默认情况下，offset是异步提交的，这个异步性导致提交到Broker的offset与Consumer 实际消费的消息并不一致。这个不一致的差值就是可能会重复消费的消息。

        同步提交：consumer提交了其消费完毕的一批消息的offset给broker后，需要等待broker的成功 ACK。当收到ACK后，consumer才会继续获取并消费下一批消息。在等待ACK期间，consumer 是阻塞的。

        异步提交：consumer提交了其消费完毕的一批消息的offset给broker后，不需要等待broker的成 功ACK。consumer可以直接获取并消费下一批消息。

        对于一次性读取消息的数量，需要根据具体业务场景选择一个相对均衡的是很有必要的。因为 数量过大，系统性能提升了，但产生重复消费的消息数量可能会增加；数量过小，系统性能会 下降，但被重复消费的消息数量可能会减少。

        消费突刺：由于Rebalance可能导致重复消费，如果需要重复消费的消息过多，或者因为Rebalance暂停 时间过长从而导致积压了部分消息。那么有可能会导致在Rebalance结束之后瞬间需要消费很多消息。

Rebalance产生的原因

        导致Rebalance产生的原因，无非就两个：消费者所订阅Topic的Queue数量发生变化，或消费者组中消费者的数量发生变化。

1）Queue数量发生变化的场景：

        Broker扩容或缩容

        Broker升级运维

        Broker与NameServer间的网络异常

        Queue扩容或缩容

2）消费者数量发生变化的场景：

        Consumer Group扩容或缩容

        Consumer升级运维

        Consumer与NameServer间网络异常

Rebalance过程

         在Broker中维护着多个Map集合，这些集合中动态存放着当前Topic中Queue的信息、Consumer Group 中Consumer实例的信息。一旦发现消费者所订阅的Queue数量发生变化，或消费者组中消费者的数量发生变化，立即向Consumer Group中的每个实例发出Rebalance通知。

TopicConågManager：key是topic名称，value是TopicConåg。TopicConåg中维护着该Topic中所 有Queue的数据。

ConsumerManager：key是Consumser Group Id，value是ConsumerGroupInfo。 ConsumerGroupInfo中维护着该Group中所有Consumer实例数据。

ConsumerOffsetManager：key为Topic与订阅该Topic的Group的组合,即topic@group， value是一个内层Map。内层Map的key为QueueId，内层Map的value为该Queue的消费进度 offset。

        Consumer实例在接收到通知后会采用Queue分配算法自己获取到相应的Queue，即由Consumer实例自主进行Rebalance。

4 Queue分配算法

        一个Topic中的Queue只能由Consumer Group中的一个Consumer进行消费，而一个Consumer可以同时 消费多个Queue中的消息。那么Queue与Consumer间的配对关系是如何确定的，即Queue要分配给哪 个Consumer进行消费，也是有算法策略的。常见的有四种策略。这些策略是通过在创建Consumer时的构造器传进去的。

1.平均分配策略

        该算法是要根据avg = QueueCount / ConsumerCount 的计算结果进行分配的。如果能够整除， 则按顺序将avg个Queue逐个分配Consumer；如果不能整除，则将多余出的Queue按照Consumer顺序 逐个分配。

该算法即，先计算好每个Consumer应该分得几个Queue，然后再依次将这些数量的Queue逐个 分配个Consumer。

2.环形平均策略

环形平均算法是指，根据消费者的顺序，依次在由queue队列组成的环形图中逐个分配。

该算法不用事先计算每个Consumer需要分配几个Queue，直接一个一个分即可

3.一致性hash策略

        该算法会将consumer的hash值作为Node节点存放到hash环上，然后将queue的hash值也放到hash环 上，通过顺时针方向，距离queue最近的那个consumer就是该queue要分配的consumer。

该算法存在的问题：分配不均

4.同机房策略

        该算法会根据queue的部署机房位置和consumer的位置，过滤出当前consumer相同机房的queue。然 后按照平均分配策略或环形平均策略对同机房queue进行分配。如果没有同机房queue，则按照平均分 配策略或环形平均策略对所有queue进行分配。

区别：

一致性hash算法存在的问题：

        两种平均分配策略的分配效率较高，一致性hash策略的较低。因为一致性hash算法较复杂。另外，一 致性hash策略分配的结果也很大可能上存在不平均的情况。

一致性hash算法存在的意义：

        其可以有效减少由于消费者组扩容或缩容所带来的大量的Rebalance。

一致性hash算法的应用场景：

        Consumer数量变化较频繁的场景。

5 至少一次原则

RocketMQ有一个原则：每条消息必须要被成功消费一次。

那么什么是成功消费呢？

        Consumer在消费完消息后会向其消费进度记录器提交其消费消息的offset， offset被成功记录到记录器中，那么这条消费就被成功消费了。

什么是消费进度记录器？ 对于广播消费模式来说，Consumer本身就是消费进度记录器。 对于集群消费模式来说，Broker是消费进度记录器。

五、订阅关系的一致性

订阅关系的一致性:

        同一个消费者组（Group ID相同）下所有Consumer实例所订阅的Topic与Tag及对消息的处理逻辑必须完全一致。否则，消息消费的逻辑就会混乱，甚至导致消息丢失。

1 正确订阅关系

        多个消费者组订阅了多个Topic，并且每个消费者组里的多个消费者实例的订阅关系保持了一致。

2 错误订阅关系

        一个消费者组订阅了多个Topic，但是该消费者组里的多个Consumer实例的订阅关系并没有保持一致。

六、offset管理

offset指的是Consumer的消费进度offset。

        消费进度offset是用来记录每个Queue的不同消费组的消费进度的。根据消费进度记录器的不同，可以 分为两种模式：本地模式和远程模式。

1 offset本地管理模式

        当消费模式为广播消费时，offset使用本地模式存储。因为每条消息会被所有的消费者消费，每个消费 者管理自己的消费进度，各个消费者之间不存在消费进度的交集。

        Consumer在广播消费模式下offset相关数据以json的形式持久化到Consumer本地磁盘文件中，默认文 件路径为当前用户主目录下的.rocketmq_offsets/${clientId}/${group}/Offsets.json 。 其中${clientId}为当前消费者id，默认为ip@DEFAULT；${group}为消费者组名称。

2 offset远程管理模式

        当消费模式为集群消费时，offset使用远程模式管理。因为所有Cosnumer实例对消息采用的是均衡消 费，所有Consumer共享Queue的消费进度。

        Consumer在集群消费模式下offset相关数据以json的形式持久化到Broker磁盘文件中，文件路径为当前 用户主目录下的store/config/consumerOffset.json 。

        Broker启动时会加载这个文件，并写入到一个双层Map（ConsumerOffsetManager）。外层map的key 为topic@group，value为内层map。内层map的key为queueId，value为offset。当发生Rebalance时， 新的Consumer会从该Map中获取到相应的数据来继续消费。

        集群模式下offset采用远程管理模式，主要是为了保证Rebalance机制。

3 offset用途

        消费者是如何从最开始持续消费消息的？消费者要消费的第一条消息的起始位置是用户自己通过 consumer.setConsumeFromWhere()方法指定的。

        在Consumer启动后，其要消费的第一条消息的起始位置常用的有三种，这三种位置可以通过枚举类型常量设置。这个枚举类型为ConsumeFromWhere。

CONSUME_FROM_LAST_OFFSET：从queue的当前最后一条消息开始消费 CONSUME_FROM_FIRST_OFFSET：从queue的第一条消息开始消费 CONSUME_FROM_TIMESTAMP：从指定的具体时间戳位置的消息开始消费。这个具体时间戳 是通过另外一个语句指定的 。

consumer.setConsumeTimestamp(“20210701080000”) yyyyMMddHHmmss

        当消费完一批消息后，Consumer会提交其消费进度offset给Broker，Broker在收到消费进度后会将其更 新到那个双层Map（ConsumerOffsetManager）及consumerOffset.json文件中，然后向该Consumer进 行ACK，而ACK内容中包含三项数据：当前消费队列的最小offset（minOffset）、最大 offset（maxOffset）、及下次消费的起始offset（nextBeginOffset）。

4 重试队列

        当rocketMQ对消息的消费出现异常时，会将发生异常的消息的offset提交到Broker中的重试队列。系统 在发生消息消费异常时会为当前的topic@group创建一个重试队列，该队列以%RETRY%开头，到达重试时间后进行消费重试。 

5 offset的同步提交与异步提交

集群消费模式下，Consumer消费完消息后会向Broker提交消费进度offset，其提交方式分为两种：

        同步提交：消费者在消费完一批消息后会向broker提交这些消息的offset，然后等待broker的成功响 应。若在等待超时之前收到了成功响应，则继续读取下一批消息进行消费（从ACK中获取 nextBeginOffset）。若没有收到响应，则会重新提交，直到获取到响应。而在这个等待过程中，消费 者是阻塞的。其严重影响了消费者的吞吐量。

        异步提交：消费者在消费完一批消息后向broker提交offset，但无需等待Broker的成功响应，可以继续 读取并消费下一批消息。这种方式增加了消费者的吞吐量。但需要注意，broker在收到提交的offset 后，还是会向消费者进行响应的。可能还没有收到ACK，此时Consumer会从Broker中直接获取 nextBeginOffset。

七、消费幂等

1 什么是消费幂等

        当出现消费者对某条消息重复消费的情况时，重复消费的结果与消费一次的结果是相同的，并且多次消 费并未对业务系统产生任何负面影响，那么这个消费过程就是消费幂等的。        

幂等：若某操作执行多次与执行一次对系统产生的影响是相同的，则称该操作是幂等的。

        在互联网应用中，尤其在网络不稳定的情况下，消息很有可能会出现重复发送或重复消费。如果重复的 消息可能会影响业务处理，那么就应该对消息做幂等处理。

2 消息重复的场景分析

什么情况下可能会出现消息被重复消费呢？最常见的有以下三种情况：

发送时消息重复

        当一条消息已被成功发送到Broker并完成持久化，此时出现了网络闪断，从而导致Broker对Producer应 答失败。 如果此时Producer意识到消息发送失败并尝试再次发送消息，此时Broker中就可能会出现两 条内容相同并且Message ID也相同的消息，那么后续Consumer就一定会消费两次该消息。

消费时消息重复

        消息已投递到Consumer并完成业务处理，当Consumer给Broker反馈应答时网络闪断，Broker没有接收 到消费成功响应。为了保证消息至少被消费一次的原则，Broker将在网络恢复后再次尝试投递之前已 被处理过的消息。此时消费者就会收到与之前处理过的内容相同、Message ID也相同的消息。

Rebalance时消息重复

        当Consumer Group中的Consumer数量发生变化时，或其订阅的Topic的Queue数量发生变化时，会触 发Rebalance，此时Consumer可能会收到曾经被消费过的消息。

3 通用解决方案

两要素

        幂等解决方案的设计中涉及到两项要素：幂等令牌，与唯一性处理。只要充分利用好这两要素，就可以 设计出好的幂等解决方案。

• 幂等令牌：是生产者和消费者两者中的既定协议，通常指具备唯⼀业务标识的字符串。例如，订 单号、流水号。一般由Producer随着消息一同发送来的。
• 唯一性处理：服务端通过采用⼀定的算法策略，保证同⼀个业务逻辑不会被重复执行成功多次。 例如，对同一笔订单的多次支付操作，只会成功一次。

解决方案

     对于常见的系统，幂等性操作的通用性解决方案是：

        1. 首先通过缓存去重。在缓存中如果已经存在了某幂等令牌，则说明本次操作是重复性操作；若缓 存没有命中，则进入下一步。

        2. 在唯一性处理之前，先在数据库中查询幂等令牌作为索引的数据是否存在。若存在，则说明本次 操作为重复性操作；若不存在，则进入下一步。

        3. 在同一事务中完成三项操作：唯一性处理后，将幂等令牌写入到缓存，并将幂等令牌作为唯一索 引的数据写入到DB中。

第1步已经判断过是否是重复性操作了，为什么第2步还要再次判断？能够进入第2步，说明已经 不是重复操作了，第2次判断是否重复？

当然不重复。一般缓存中的数据是具有有效期的。缓存中数据的有效期一旦过期，就是发生缓 存穿透，使请求直接就到达了DBMS。

解决方案举例

以支付场景为例：

        1. 当支付请求到达后，首先在Redis缓存中却获取key为支付流水号的缓存value。若value不空，则 说明本次支付是重复操作，业务系统直接返回调用侧重复支付标识；若value为空，则进入下一步 操作

        2. 到DBMS中根据支付流水号查询是否存在相应实例。若存在，则说明本次支付是重复操作，业务 系统直接返回调用侧重复支付标识；若不存在，则说明本次操作是首次操作，进入下一步完成唯 一性处理

        3. 在分布式事务中完成三项操作：

• 完成支付任务
•  将当前支付流水号作为key，任意字符串作为value，通过set(key, value, expireTime)将数 据写入到Redis缓存
• 将当前支付流水号作为主键，与其它相关数据共同写入到DBMS

4 消费幂等的实现

        消费幂等的解决方案很简单：为消息指定不会重复的唯一标识。因为Message ID有可能出现重复的情 况，所以真正安全的幂等处理，不建议以Message ID作为处理依据。最好的方式是以业务唯一标识作为 幂等处理的关键依据，而业务的唯一标识可以通过消息Key设置。

以支付场景为例，可以将消息的Key设置为订单号，作为幂等处理的依据。具体代码示例如下：

Message message = new Message();

message.setKey("ORDERID_100");

SendResult sendResult = producer.send(message);

消费者收到消息时可以根据消息的Key即订单号来实现消费幂等：

consumer.registerMessageListener(new MessageListenerConcurrently() {
    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,ConsumeConcurrentlyContext  context) {
        for (MessageExt msg : msgs) {
            String key = msg.getKeys();
            // 根据业务唯一标识Key做幂等处理
            // ……
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
});

RocketMQ能够保证消息不丢失，但不能保证消息不重复。

八、消息堆积与消费延迟

1 概念

        消息处理流程中，如果Consumer的消费速度跟不上Producer的发送速度，MQ中未处理的消息会越来 越多（进的多出的少），这部分消息就被称为堆积消息。消息出现堆积进而会造成消息的消费延迟。 以下场景需要重点关注消息堆积和消费延迟问题：

• 业务系统上下游能力不匹配造成的持续堆积，且无法自行恢复。
• 业务系统对消息的消费实时性要求较高，即使是短暂的堆积造成的消费延迟也无法接受。

2 产生原因分析

Consumer使用长轮询Pull模式消费消息时，分为以下两个阶段：

消息拉取

        Consumer通过长轮询Pull模式批量拉取的方式从服务端获取消息，将拉取到的消息缓存到本地缓冲队 列中。对于拉取式消费，在内网环境下会有很高的吞吐量，所以这一阶段一般不会成为消息堆积的瓶颈。

一个单线程单分区的低规格主机(Consumer，4C8G)，其可达到几万的TPS。如果是多个分区多 个线程，则可以轻松达到几十万的TPS。

消息消费

        Consumer将本地缓存的消息提交到消费线程中，使用业务消费逻辑对消息进行处理，处理完毕后获取 到一个结果。这是真正的消息消费过程。此时Consumer的消费能力就完全依赖于消息的消费耗时和消费并发度了。如果由于业务处理逻辑复杂等原因，导致处理单条消息的耗时较长，则整体的消息吞吐 量肯定不会高，此时就会导致Consumer本地缓冲队列达到上限，停止从服务端拉取消息。

结论

        消息堆积的主要瓶颈在于客户端的消费能力，而消费能力由消费耗时和消费并发度决定。注意，消费 耗时的优先级要高于消费并发度。即在保证了消费耗时的合理性前提下，再考虑消费并发度问题。

3 消费耗时

        影响消息处理时长的主要因素是代码逻辑。而代码逻辑中可能会影响处理时长代码主要有两种类型： CPU内部计算型代码和外部I/O操作型代码。

        通常情况下代码中如果没有复杂的递归和循环的话，内部计算耗时相对外部I/O操作来说几乎可以忽略。所以外部IO型代码是影响消息处理时长的主要症结所在。

外部IO操作型代码举例：

• 读写外部数据库，例如对远程MySQL的访问
• 读写外部缓存系统，例如对远程Redis的访问
• 下游系统调用，例如Dubbo的RPC远程调用，Spring Cloud的对下游系统的Http接口调用

关于下游系统调用逻辑需要进行提前梳理，掌握每个调用操作预期的耗时，这样做是为了能够 判断消费逻辑中IO操作的耗时是否合理。通常消息堆积是由于下游系统出现了服务异常或达到 了DBMS容量限制，导致消费耗时增加。

服务异常，并不仅仅是系统中出现的类似500这样的代码错误，而可能是更加隐蔽的问题。例 如，网络带宽问题。

达到了DBMS容量限制，其也会引发消息的消费耗时增加。

4 消费并发度

        一般情况下，消费者端的消费并发度由单节点线程数和节点数量共同决定，其值为单节点线程数*节点数量。不过，通常需要优先调整单节点的线程数，若单机硬件资源达到了上限，则需要通过横向扩展 来提高消费并发度。

单节点线程数，即单个Consumer所包含的线程数量

节点数量，即Consumer Group所包含的Consumer数量

对于普通消息、延时消息及事务消息，并发度计算都是单节点线程数*节点数量。但对于顺序 消息则是不同的。顺序消息的消费并发度等于Topic的Queue分区数量。

1）全局顺序消息：该类型消息的Topic只有一个Queue分区。其可以保证该Topic的所有消息被 顺序消费。为了保证这个全局顺序性，Consumer Group中在同一时刻只能有一个Consumer的一 个线程进行消费。所以其并发度为1。

2）分区顺序消息：该类型消息的Topic有多个Queue分区。其仅可以保证该Topic的每个Queue 分区中的消息被顺序消费，不能保证整个Topic中消息的顺序消费。为了保证这个分区顺序性， 每个Queue分区中的消息在Consumer Group中的同一时刻只能有一个Consumer的一个线程进行 消费。即，在同一时刻最多会出现多个Queue分蘖有多个Consumer的多个线程并行消费。所以 其并发度为Topic的分区数量。

5 单机线程数计算

        对于一台主机中线程池中线程数的设置需要谨慎，不能盲目直接调大线程数，设置过大的线程数反而会 带来大量的线程切换的开销。理想环境下单节点的最优线程数计算模型为：C *（T1 + T2）/ T1。

• C：CPU内核数
• T1：CPU内部逻辑计算耗时
• T2：外部IO操作耗时

最优线程数 = C *（T1 + T2）/ T1 = C * T1/T1 + C * T2/T1 = C + C * T2/T1

注意，该计算出的数值是理想状态下的理论数据，在生产环境中，不建议直接使用。而是根据 当前环境，先设置一个比该值小的数值然后观察其压测效果，然后再根据效果逐步调大线程 数，直至找到在该环境中性能最佳时的值。

6 如何避免

        为了避免在业务使用时出现非预期的消息堆积和消费延迟问题，需要在前期设计阶段对整个业务逻辑进 行完善的排查和梳理。其中最重要的就是梳理消息的消费耗时和设置消息消费的并发度。

梳理消息的消费耗时

        通过压测获取消息的消费耗时，并对耗时较高的操作的代码逻辑进行分析。梳理消息的消费耗时需要关注以下信息：

• 消息消费逻辑的计算复杂度是否过高，代码是否存在无限循环和递归等缺陷。
• 消息消费逻辑中的I/O操作是否是必须的，能否用本地缓存等方案规避。
• 消费逻辑中的复杂耗时的操作是否可以做异步化处理。如果可以，是否会造成逻辑错乱。

设置消费并发度

对于消息消费并发度的计算，可以通过以下两步实施：

• 逐步调大单个Consumer节点的线程数，并观测节点的系统指标，得到单个节点最优的消费线程数 和消息吞吐量。
• 根据上下游链路的流量峰值计算出需要设置的节点数

节点数 = 流量峰值 / 单个节点消息吞吐量

九、消息的清理

        消息被消费过后会被清理掉吗？不会的。

        消息是被顺序存储在commitlog文件的，且消息大小不定长，所以消息的清理是不可能以消息为单位进行清理的，而是以commitlog文件为单位进行清理的。否则会急剧下降清理效率，并实现逻辑复杂。

        commitlog文件存在一个过期时间，默认为72小时，即三天。除了用户手动清理外，在以下情况下也 会被自动清理，无论文件中的消息是否被消费过：

• 文件过期，且到达清理时间点（默认为凌晨4点）后，自动清理过期文件
• 文件过期，且磁盘空间占用率已达过期清理警戒线（默认75%）后，无论是否达到清理时间点， 都会自动清理过期文件
• 磁盘占用率达到清理警戒线（默认85%）后，开始按照设定好的规则清理文件，无论是否过期。 默认会从最老的文件开始清理
• 磁盘占用率达到系统危险警戒线（默认90%）后，Broker将拒绝消息写入

需要注意以下几点：

        1）对于RocketMQ系统来说，删除一个1G大小的文件，是一个压力巨大的IO操作。在删除过程 中，系统性能会骤然下降。所以，其默认清理时间点为凌晨4点，访问量最小的时间。也正因如 果，我们要保障磁盘空间的空闲率，不要使系统出现在其它时间点删除commitlog文件的情况。

        2）官方建议RocketMQ服务的Linux文件系统采用ext4。因为对于文件删除操作，ext4要比ext3性 能更好 

十、RokectMQ防消息丢失

1.RokectMQ消息什么情况下会丢失？

        MQ的消息生成到消费主要经历三个阶段：MQ消息生产、RocketMQ Broker存储消息、消费者消息对应的消息。

        消息丢失主要会发生在下面几个地方：发送、存储、消费消息的三个阶段。

2.如何解决RokectMQ消息丢失：

        解决消息丢失从消息丢失的地方入手。

3.消息生产防止消息丢失：

        RocketMQ消息生产方式有三种：同步发送消息、异步发送消息、One-Way发送消息。不同的发送方式试用不同的场景：

        同步发送消息： 重要的通知(订单状态的更新)、短信系统。

        异步发送消息： 通常用于响应时间敏感的业务场景。

        One-way： 主要用于对可靠性要求不高的场景，在金融的场景下不适用。一般是用于日志收集。

SendResult定义说明：

 SEND_OK：

                消息发送成功。要注意的是消息发送成功也不意味着它是可靠的。要确保不会丢失任何消息，还应启用同步Master服务器或同步刷盘，即SYNC_MASTER或SYNC_FLUSH。        

FLUSH_DISK_TIMEOUT：

                消息发送成功但是服务器刷盘超时。此时消息已经进入服务器队列（内存），只有服务器宕机，消息才会丢失。消息存储配置参数中可以设置刷盘方式和同步刷盘时间长度，如果Broker服务器设置了刷盘方式为同步刷盘，即FlushDiskType=SYNC_FLUSH（默认为异步刷盘方式），当Broker服务器未在同步刷盘时间内（默认为5s）完成刷盘，则将返回该状态——刷盘超时。

FLUSH_SLAVE_TIMEOUT：

                消息发送成功，但是服务器同步到Slave时超时。此时消息已经进入服务器队列，只有服务器宕机，消息才会丢失。如果Broker服务器的角色是同步Master，即SYNC_MASTER（默认是异步Master即ASYNC_MASTER），并且从Broker服务器未在同步刷盘时间（默认为5秒）内完成与主服务器的同步，则将返回该状态——数据同步到Slave服务器超时。

SLAVE_NOT_AVAILABLE：

        消息发送成功，但是此时Slave不可用。如果Broker服务器的角色是同步Master，即SYNC_MASTER（默认是异步Master服务器即ASYNC_MASTER），但没有配置slave Broker服务器，则将返回该状态——无Slave服务器可用。

4.RocketMQ Broker防丢失消息：

各种模式遵循原则：

        异步刷盘丢失少量消息，同步刷盘一条不丢

5.消费端处理消息

/**
 * 消费重试配置方式
 *
 *
 * 集群消费方式下，消息消费失败后若希望消费重试，则需要在消息监听器接口的实现中明确进行如下三
 * 种方式之一的配置：
 *  方式1：返回ConsumeConcurrentlyStatus.RECONSUME_LATER（推荐）
 *  方式2：返回Null
 *  方式3：抛出异常
 *
 */
public class RetryConsumer {

    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg");
        consumer.setNamesrvAddr("127.0.0.1:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("someTopic", "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {

            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
               try {
                   
               } catch (Throwable e) {
                   // 消费失败
                   return ConsumeConcurrentlyStatus.RECONSUME_LATER;
//                   return null;
//                   throw   new RuntimeException("消费异常");
               }

                // 返回消费状态：消费成功
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 开启消费者消费
        consumer.start();
        System.out.printf("Consumer Started.%n");
    }
}

总结
        MQ消息的丢失主要发生在发送、存储、消费消息的三个阶段，所以需要防止消息丢失也要从这三个方面着手。

        发送消息使用同步或者异步的方式，然后根据返回的 SendResult 来判断是否发送成功
        Broker的刷盘方式配置成同步刷盘
        消息消息失败根据业务需要来判断是否需要重新消费消息。

十一、与Kafka的对比

1. Kafka

• Topic A共有2个分区，每个分区有两个副本(一个Leader一个Follower)，分别分布在 Broker0 和 Broker1 上；
• Broker0/Broker1每台机器既是Leader，也是Follower。具体来说，比如机器Broker0对于Partition0来说是Master，对于Partition1来说又是Follower；

消息存储：

• Kafka使用异步刷盘方式，异步Replication
•  ​​​​​​​Kafka单机写入TPS约在百万条/秒，消息大小10个字节
•  Kafka单机超过64个队列/分区，Load会发生明显的飙高现象，队列越多，load越高，发送消息响应时间变长
• Kafka使用短轮询方式，实时性取决于轮询间隔时间
• Kafka消费失败不支持重试
• Kafka支持消息顺序，但是一台Broker宕机后，就会产生消息乱序
• Kafka不支持定时消息
• Kafka不支持消息查询
• Kafka理论上可以按照Offset来回溯消息

2. RocketMQ

• Broker集群中有Broker0和Broker1为Leader，每个Broker有两个Slave，共6台机器；
• Master0/Slave0_0/Slave0_1/Master1/Slave1_0/Slave1_1每台机器通过配置固定只能充当Master或Slave；
• TopicA在Broker0上有两个队列，在Broker1上有两个队列；

消息存储：

•  RocketMQ支持异步实时刷盘，同步刷盘，同步Replication，异步Replication
• RocketMQ单机写入TPS单实例约7万条/秒，单机部署3个Broker，可以跑到最高12万条/秒，消息大小10个字节
• RocketMQ单机支持最高5万个队列，Load不会发生明显变化
• RocketMQ使用长轮询，同Push方式实时性一致，消息的投递延时通常在几个毫秒。
• RocketMQ消费失败支持定时重试，每次重试间隔时间顺延
• RocketMQ支持严格的消息顺序，在顺序消息场景下，一台Broker宕机后，发送消息会失败，但是不会乱序

• RocketMQ支持两类定时消息
  • 开源版本RocketMQ仅支持定时Level
  • 阿里云ONS支持定时Level，以及指定的毫秒级别的延时时间

• RocketMQ支持根据Message Id查询消息，也支持根据消息内容查询消息（发送消息时指定一个Message Key，任意字符串，例如指定为订单Id）

• RocketMQ支持按照时间来回溯消息，精度毫秒，例如从一天之前的某时某分某秒开始重新消费消息

注意：

        MQ消息的丢失主要发生在发送、存储、消费消息的三个阶段，所以需要防止消息丢失也要从这三个方面着手。

• 发送消息使用同步或者异步的方式，然后根据返回的 SendResult 来判断是否发送成功
• Broker的刷盘方式配置成同步刷盘
• 消息消息失败根据业务需要来判断是否需要重新消费消息。