RocketMQ 如何保证消息不丢失？如何快速处理积压消息？

1. 哪些环节会有丢消息的可能？
   关于 MQ，有一个问题是无法避免的，就是怎么保证消息不丢失，这个问题是所有 MQ 都需要面对的一个共性问题。大致的解决思路都是一致的，首先要找到哪些环节会有丢消息的可能，来看一个 MQ 的通用架构

其中，1，2，4 三个场景分别是：发消息、消息主从同步、消费消息 他们都是跨网络的，而跨网络就肯定会有丢消息的可能。

    然后关于 3 这个环节，通常 MQ 存盘时都会先写入操作系统的缓存 page cache 中，然后再由操作系统异步的将消息写入硬盘，这个中间有个时间差，就可能会造成消息丢失。比如服务挂了，缓存中还没有来得及写入硬盘的消息就会丢失。还有一个点：MQ 整个服务挂了，也会丢消息！

综上所述：MQ 一般会在以下几种情景中丢失消息：

生产者发消息到 Broker 时
Broker 主从同步
Broker 存储消息时
消费者消费消息时
整个 MQ 服务宕机
下面看一下 RocketMQ 如何解决这个问题的

1. 消息生产阶段如何保证消息不丢失
   

2.1 同步发送
Produce 有三种发消息的方式

同步发送
异步发送
单向发送
由于同步和异步方式均需要 Broker 返回确认信息，单向发送只管发，不需要 Broker 返回确认信息，所以单向发送并不知道消息是不是发送成功，单向发送不能保证消息不丢失。

    produce 要想发消息时保证消息不丢失，可以采用同步发送的方式去发消息，send 消息方法只要不抛出异常，就代表发送成功。发送成功会有多个状态，以下对每个状态进行说明：

SEND_OK：消息发送成功，Broker 刷盘、主从同步成功
FLUSH_DISK_TIMEOUT：消息发送成功，但是服务器同步刷盘（默认为异步刷盘）超时（默认超时时间 5 秒）
FlUSH_SLAVE_TIMEOUT：消息发送成功，但是服务器同步复制（默认为异步复制）到 Slave 时超时（默认超时时间 5 秒）
SLAVE_NOT_AVAILABLE：Broker 从节点不存在
注意：同步发送只要返回以上四种状态，就代表该消息在生产阶段消息正确的投递到了 RocketMq，没有丢失。但也只是消息投递过程没有丢失，并不代表存储阶段消息不丢失，当出现超时或失败状态时，则会触发默认的 2 次重试。从生产者角度看，如果消息未能正确的存储在 MQ 中，或者消费者未能正确的消费到这条消息，都是消息丢失。

    如果业务要求严格，为保证消息的成功投递和保存，我们可以只取 SEND_OK 标识消息发送成功，把失败的消息记录到数据库，并启动一个定时任务，扫描发送失败的消息，重新发送 5 次，直到成功或者发送 5 次后发送邮件或短信通知人工介入处理。如下代码所示：

    log.info(发送消息mq  key  发送中........)         //用数据库记录日志：  插入本地消息表：消息：key, 消息状态：发送中
    try{
        //发消息
         SendResult   sendResult = send(msg)
       // 状态为SEND_OK标识发送成功
       if(sendResult .getSendStatus() == SEND_OK){
              log.info(发送消息mq  key  发送sucess........)  //用数据库记录日志：  更新本地消息表：消息：key, 消息状态：发送成功                     
       }else{
           log.info(发送消息mq  key  发送fail........)  //用数据库记录日志： 更新本地消息表：消息：key, 消息状态：发送失败
      }

2.2 采用事务消息
这个结论比较容易理解，因为 RocketMQ 的事务消息机制就是为了保证零丢失来设计的，并且经过阿里的验证，肯定是非常靠谱的。点击查看事务消息详情！！！ 我们以最常见的电商订单场景为例，来简单分析下事务消息机制如何保证消息不丢失。我们看下下面这个流程图：

1、为什么要发送个 half 消息？有什么用？

    我们通常是会在订单系统中先完成下单，再发送消息给 MQ，如果发送到 MQ 时，发现 MQ 宕机了，就会非常尴尬了。而这个 half 消息是在订单系统进行下单操作前发送，并且对下游服务的消费者是不可见的。那这个消息的作用更多的体现在确认 RocketMQ 的服务是否正常。相当于嗅探下 RocketMQ 服务是否正常，并且通知 RocketMQ，我马上就要发一个很重要的消息了，你做好准备。

2.half 消息如果写入失败了怎么办？

    如果 half 消息如果写入失败，我们就可以认为 MQ 的服务是有问题的，这时，就不能通知下游服务了。我们可以在下单时给订单一个状态标记入库，然后等待 MQ 服务正常后再重新下单通知下游服务。可以使用定时任务实现！

1. 本地事务中订单系统写数据库失败了怎么办？

    如果没有使用事务消息，我们只能判断下单失败，抛出了异常，那就不往 MQ 发消息了，这样至少保证不会对下游服务进行错误的通知。但是这样的话，如果过一段时间可以正常下单了，那么这个异常单就丢失了，需要用户人为的重新点击。当然，也可以设计另外的补偿机制，例如：将这次异常下单数据缓存起来，再启动一个线程定时尝试往数据库写，省去用户重复点击的操作。

可以有一种更优雅的方案，就是使用事务消息机制：

如果下单时，写数据库失败 (可能是数据库崩了，需要等一段时间才能恢复)。那我们可以另外找个地方把订单消息先缓存起来 (Redis、文本或者其他方式)，然后给 RocketMQ 返回一个 UNKNOWN 状态。
这样 RocketMQ 就会过一段时间来回查事务状态。我们就可以在回查事务状态时尝试把已保存至 redis 缓存的订单数据再次写入数据库，如果数据库这时候已经恢复了，那就能完整正常的下单，再继续后面的业务，如果没有恢复，会进行多次回查，直到数据库恢复，这样这个订单的消息就不会因为数据库临时崩了而丢失。如果多次回查数据库依旧没有恢复，再采用其他补偿机制。
4. 下单成功后如何优雅的等待支付成功？

    在订单场景下，通常会要求下单完成后，客户在一定时间内，例如 10 分钟，内完成订单支付，支付完成后才会通知下游服务进行进一步的营销补偿。10 分钟内未支付，该订单超时！

如果不使用事务消息，可以有以下方案：

最简单的方式是启动一个定时任务，每隔一段时间扫描订单表，比对未支付的订单的下单时间，将超过时间的订单回收。这种方式显然是有很大问题的，需要定时扫描很庞大的一个订单信息，这对系统是个不小的压力。
使用 RocketMQ 的延迟消息机制，在订单生成时，往 MQ 发一个延迟 1 分钟的消息，消息内容为订单号，消费到这个消息后去检查订单的支付状态，如果订单已经支付，就往下游发送通知。如果没有支付，就再发一个延迟 1 分钟的消息。最终在第十个消息时把订单回收。这个方案就不用对全部的订单表进行扫描，而只需要每次处理一个单独的延时消息。
如果使用事务消息，可以更优雅的处理订单超时问题

    可以用事务消息的状态回查机制来替代定时的任务。在下单时，给 Broker 返回一个 UNKNOWN 的未知状态。而在状态回查的方法中去查询订单的支付状态。这样整个业务逻辑就会简单很多。我们只需要配置 RocketMQ 中的事务消息回查次数 (默认 15 次) 和事务回查间隔时间 (messageDelayLevel)，就可以更优雅的完成这个支付状态检查的需求。

1. 使用事务消息可以解决 生产者 >> Broker >> 消费者 之间的分布式事务吗？

    整体来说，在订单这个场景下，消息不丢失的问题实际上就还是转化成了下单这个业务与下游服务的业务的分布式事务一致性问题。而事务一致性问题一直以来都是一个非常复杂的问题。
    RocketMQ 的事务消息机制，实际上只保证了整个事务消息的一半，他保证的是订单系统下单和发消息这两个事件的事务一致性，而对下游服务的事务并没有保证。但是即便如此，也是分布式事务的一个很好的降级方案。目前来看，也是业内最好的降级方案。

1. Broker 如何保证接收到的消息不会丢失
   使用同步刷盘机制：

    同步刷盘机制，只有在消息真正持久化至磁盘后，RocketMQ 的 Broker 端才会真正地返回给 Producer 端一个成功的 ACK 响应，保证了消息可靠性，但影响了性能。异步刷盘则能够充分利用 OS 的 PageCache 的优势，只要消息写入 PageCache 即可将成功的 ACK 返回给 Producer 端，消息刷盘采用后台异步线程提交的方式进行，提高了 MQ 的性能和吞吐量，但是可能会丢消息。点击查看配置方式

使用同步复制机制：

    同步复制是等 Master 和 Slave 都写入消息成功后才反馈给客户端写入成功的状态。在同步复制下，如果 Master 节点故障，Slave 上有全部的数据备份，这样容易恢复数据。但是同步复制会增大数据写入的延迟，降低系统的吞吐量。异步复制是只要 master 写入消息成功，就反馈给客户端写入成功的状态。速度快，同样可能丢消息！

1. 消费者如何确保拉取到的消息被成功消费？
   正常情况下，rocketMq 拉取消息后，执行业务逻辑。一旦执行成功，将会返回一个 ACK 响应给 Broker，这时 MQ 就会修改 offset，将该消息标记为已消费，不再往其他消费者推送消息。

    如果出现消费超时 (默认15分钟)、拉取消息后消费者服务宕机等消费失败的情况，此时的 Broker 由于没有等到消费者返回的 ACK，会向同一个消费者组中的其他消费者间隔性的重发消息，直到消息返回成功（默认是重复发送 16 次，若 16 次还是没有消费成功，那么该消息会转移到死信队列，人工处理或是单独写服务处理这些死信消息）
    在 Broker 的这种重新推送机制下，正常同步消费是不会丢消息的，但是异步消费就不一定，比如下面这种情况：

 DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                new Thread(){
                    public void run(){
                        //处理业务逻辑
                        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                    }
                };
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

    在监听器中，新启动了一个线程，异步处理业务逻辑。监听器返回了消费成功 ConsumeConcurrentlyStatus.CONSUME_SUCCESS ，但是异步处理线程业务却失败了，那么这个消息并没有被消费，但是在 Broker 中已经标记为消费过了，导致消息丢失，只不过这里的丢失是业务代码处理不合理造成的。

所以，为了保证消费的可靠性，消费端尽量不要使用异步消费机制，如果你一定要使用异步，那么就记录消费失败的日志入库，开启其他线程重新消费！

消费幂等性如何保证？

由于存在重试机制（消息重新发送），在消费时需要做幂等性校验，防止重复消费。解决方案如下：

第一次消费时，可以把消息的唯一 ID，或者业务 id（orderId）存在 redis 中（key = orderId, value = 1）
如果发生了重试，先去 redis 中查一下该唯一 ID 是否存在，如果存在，则为重复消费！不再处理

1. Broker 宕机后如何防止消息丢失？
   当 NameServer 全部挂了 (注意是全部，只要有一个存活就是正常)，或者 Broker 宕机了。在这种情况下，RocketMQ 相当于整个服务都不可用了，那他本身肯定无法给我们保证消息不丢失了。只能自己设计一个降级方案来处理这个问题了。

    例如在订单系统中，如果多次尝试发送 RocketMQ 不成功，那就只能另外找给地方 (Redis、文件或者内存等) 把订单消息缓存下来，然后起一个线程定时的扫描这些失败的订单消息，尝试往 RocketMQ 发送。这样等 RocketMQ 的服务恢复过来后，就能第一时间把这些消息重新发送出去。整个这套降级的机制，在大型互联网项目中，都是必须要有的。

1. 消息零丢失方案总结
   综上所述：RocketMQ 处理消息零丢失的方案要考虑以下几点：

生产者使用同步发送，或者发送事务消息
Broker 配置同步刷盘 + 同步复制
消费者不要使用异步消费
整个 MQ 挂了之后准备降级方案
这套方案在应对绝对要求零丢失的场景下，不失为一个很好的参考，那这套方案是不是就很完美呢？其实很明显，这整套的消息零丢失方案，在各个环节都大量的降低了系统的处理性能以及吞吐量。在很多场景下，这套方案带来的性能损失的代价可能远远大于部分消息丢失的代价。

    所以，我们在设计 RocketMQ 使用方案时，要根据实际的业务情况来考虑。例如，如果针对所有服务器都在同一个机房的场景，完全可以把 Broker 配置成异步刷盘来提升吞吐量。而在有些对消息可靠性要求没有那么高的场景，在生产者端就可以采用其他一些更简单的方案来提升吞吐，而采用定时对账、补偿的机制来提高消息的可靠性。而如果消费者不需要进行消息存盘，那使用异步消费的机制带来的性能提升也是非常显著的。

1. 如何快速处理积压消息？
   
   7.1 如何确定 RocketMQ 有大量的消息积压？

    在正常情况下，使用 MQ 都会要尽量保证他的消息生产速度和消费速度整体上是平衡的，但是如果部分消费者系统出现故障，produce 还在一直发消息，消息得不到及时消费，就会造成大量的消息积累。这在一些大型的互联网项目中，消息积压的速度是相当恐怖的。所以消息积压是个需要时时关注的问题。
    对于消息积压，如果是 RocketMQ 或者 kafka 还好，他们的消息积压不会对性能造成很大的影响。而如果是 RabbitMQ 的话，那就惨了，大量的消息积压可以瞬间造成性能直线下滑。
    对于 RocketMQ 来说，有个最简单的方式来确定消息是否有积压。那就是使用 web 控制台，就能直接看到消息的积压情况。在 Web 控制台的主题页面，可以通过 Consumer 管理按钮实时看到消息的积压情况。其中差值一栏就是队列中积压的消息！

另外，也可以通过 mqadmin 指令在后台检查各个 Topic 的消息延迟情况。

还有 RocketMQ 也会在他的 ${storePathRootDir}/config 目录下落地一系列的 json 文件，也可以用来跟踪消息积压情况。

    <br />7.2 如何处理大量积压的消息？

面对大量积压的消息：

如果 Topic 下的 MessageQueue 配置得是足够多的（默认只有4个），那每个 Consumer 实际上会分配多个 MessageQueue 来进行消费。这个时候，就可以简单的通过增加 Consumer 的服务节点数量来加快消息的消费，等积压消息消费完了，再恢复成正常情况。最极限的情况是把 Consumer 的节点个数设置成跟 MessageQueue 的个数相同。但是如果此时再继续增加 Consumer 的服务节点就没有用了。
而如果 Topic 下的 MessageQueue 配置得不够多的话，那就不能用上面这种增加 Consumer 节点个数的方法了。这时怎么办呢？
这时如果要快速处理积压的消息，可以创建一个新的 Topic，并配置足够多的 MessageQueue。并紧急上线一组新的消费者，只负责搬运积压的消息，转储到新的 Topic 中，这个速度是可以很快的。然后在新的 Topic 上，就可以通过增加消费者个数来提高消费速度了。之后再根据情况恢复成正常情况。
————————————————
版权声明：本文为CSDN博主「知识分子_」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/qq_45076180/article/details/113828472