RocketMQ5.x教程-RocketMQ使用中常见的问题

RocketMQ使用中常见的问题

6.1.RocketMQ如何保证消息不丢失

我们将消息流程分为三大部分，每一部分都有可能会丢失数据。

• 生产阶段：Producer通过网络将消息发送给Broker，这个发送可能会发生丢失。比如网络延迟不可达等。
• 存储阶段：Broker肯定是先把消息放到内存的，然后根据刷盘策略持久化到硬盘中。刚收到Producer的消息，放入内存，但是异常宕机了，导致消息丢失。
• 消费阶段：消费失败。比如先提交ack再消费，处理过程中出现异常，该消息就出现了丢失。

解决方案：

• 生产阶段：使用同步发送失败重试机制；异步发送重写回调方法检查发送结果；Ack确认机制。
• 存储阶段：同步刷盘机制；集群模式采用同步复制。
• 消费阶段：正常消费处理完成才提交ACK；如果处理异常返回重试标识。

除了上述，在生产阶段与消费者阶段部分消息还需要确保消息顺序消费。

6.2.RocketMQ的消息持久化机制

RocketMQ的消息持久化机制是指将消息存储在磁盘上，以确保消息能够可靠地存储和检索。RocketMQ 的消息持久化机制涉及到以下三个角色：CommitLog、ConsumeQueue 和 IndexFile。

• CommitLog：消息真正的存储文件，所有的消息都存在 CommitLog文件中。

RocketMQ默认会将消息数据先存储到内存中的一个缓冲区，每当缓冲区中积累了一定量的消息或者一定时间后，就会将缓冲区中的消息批量写入到磁盘上的 CommitLog 文件中。消息在写入 CommitLog 文件后就可以被消费者消费了。

Commitlog文件的大小固定1G，写满之后生成新的文件，并且采用的是顺序写的方式。

• ConsumeQueue：消息消费逻辑队列，类似数据库的索引文件。

RocketMQ 中每个主题下的每个消息队列都会对应一个 ConsumeQueue。ConsumeQueue存储了消息的offset以及该offset对应的消息在CommitLog文件中的位置信息，便于消费者快速定位并消费消息。

每个ConsumeQueue文件固定由30万个固定大小20byte的数据块组成；据块的内容包括：msgPhyOffset(8byte，消息在文件中的起始位置)+msgSize(4byte，消息在文件中占用的长度)+msgTagCode(8byte，消息的tag的Hash值)。

• IndexFile：消息索引文件，主要存储消息Key与offset的对应关系，提升消息检索速度。

如果生产者在发送消息时设置了消息Key，那么RocketMQ会将消息Key值和消息的物理偏移量（offset）存储在IndexFile文件中，这样当消费者需要根据消息Key查询消息时，就可以直接在IndexFile文件中查找对应的offset，然后通过 ConsumeQueue文件快速定位并消费消息。

IndexFile文件大小固定400M，可以保存2000W个索引。

三个角色构成的消息存储结构如下：

消息存储过程：

6.3.RocketMQ如何保证消息顺序

RocketMQ架构本身是无法保证消息有序的，但是提供了相应的API保证消息有序消费。RocketMQ API利用FIFO先进先出的特性，保证生产者消息有序进入同一队列，消费者在同一队列消费就能达到消息的有序消费。

• 使用MessageQueueSelector编写有序消息生产者

有序消息生产者会按照一定的规则将消息发送到同一个队列中，从而保证同一个队列中的消息是有序的。RocketMQ 并不保证整个主题内所有队列的消息都是按照发送顺序排列的。

• 使用MessageListenerOrderly进行顺序消费与之对应的MessageListenerConcurrently并行消费（push模式）

MessageListenerOrderly是RocketMQ 专门提供的一种顺序消费的接口，它可以让消费者按照消息发送的顺序，一个一个地处理消息。这个接口支持按照消息的重试次数进行顺序消费、订单ID等作为消息键来实现顺序消费、批量消费等操作。

通过加锁的方式实现（有超时机制），一个队列同时只有一个消费者；并且存在一个定时任务，每隔一段时间就会延长锁的时间，直到整个消息队列全部消费结束。

• 消费端自己保证消息顺序消费（pull模式）
• 消费者并发消费时设置消费线程为1

RocketMQ 的消费者可以开启多个消费线程同时消费同一个队列中的消息，如果要保证消息的顺序，需要将消费线程数设置为1。这样，在同一个队列中，每个消息只会被单个消费者线程消费，从而保证消息的顺序性

rokectMQ消息模型：

6.4.RocketMQ的事务消息原理

RocketMQ 的事务消息是一种保证消息可靠性的机制。在RocketMQ中，事务消息的实现原理主要是通过两个发送阶段和一个确认阶段来实现的。

• 发送消息的预处理阶段：在发送事务消息之前，RocketMQ 会将消息的状态设置为“Preparing”，并将消息存储到消息存储库中。
• 执行本地事务：当预处理阶段完成后，消息发送者需要执行本地事务，并返回执行结果（commit 或 rollback）。
• 消息的二次确认阶段：根据本地事务的执行结果，如果是 commit，则 RocketMQ 将消息的状态设置为“Committing”；否则将消息的状态设置为“Rollback”。
• 完成事务：最后在消息的消费者消费该消息时，RocketMQ 会根据消息的状态来决定是否提交该消息。如果消息的状态是“Committing”，则直接提交该消息；否则忽略该消息。

需要注意的是，如果在消息发送的过程中出现异常或者网络故障等问题，RocketMQ 会触发消息回查机制。在回查过程中，RocketMQ 会调用消息发送方提供的回查接口来确认事务的提交状态，从而解决消息投递的不确定性。