一文理清RocketMQ顺序消费、重复消费、消息丢失问题
作者：互联网
前言
在使用消息队列时不可避免的会遇到顺序消费、重复消费、消息丢失三个问题。在一次面试字节的时候，面试官问到如何保证顺序消费，当时回答不太准确，特意此文回顾如何解决顺序消费、重复消费、消息丢失三个问题。
重复消费
解决重复消费的关键在于消费方的幂等
幂等（idempotent、idempotence）是一个数学与计算机学概念，常见于抽象代数中。在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同。
幂等函数，或幂等方法，是指可以使用相同参数重复执行，并能获得相同结果的函数。这些函数不会影响系统状态，也不用担心重复执行会对系统造成改变。
幂等操作根据场景的不同可以分为：
（1）强校验
场景：如与金钱相关的支付等关键消息，必须强校验。
基于数据库的唯一键来保证重复数据不会被插入多条。建立一个已消费消息的表，每次消费之前检查消费表中当前消费的消息是否已经存在，若存在表示消息已经被消费过直接返回。
（2）弱校验
场景：可以有小概率出现重复消费的非关键消息
基于Redis的实现：
- 使用set结构实现。每次消费前查看set中是否已经存在待消费的消息的唯一标识符，不存在则消息，存在则直接返回。
- 场景唯一标识+id作为Redis的key，并设置一定的过期时间。每次消费时检查key是否已经存在，存在则直接返回
顺序消费
RocketMQ提供两种顺序消息模式：
- 普通顺序消息
普通顺序消费模式下，消费者通过同一个消费队列收到的消息是有顺序的，不同消息队列收到的消息则可能是无顺序的。普通顺序消息在 Broker 重启情况下不会保证消息顺序性 (短暂时间) 。 - 严格顺序消息
严格顺序消息模式下，对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。严格顺序消息 即使在异常情况下也会保证消息的顺序性 。
严格顺序虽然能更好的保证消息有序，但实现它可会付出巨大的代价。如果你使用严格顺序模式，Broker 集群中只要有一台机器不可用，则整个集群都不可用。
一般而言，我们的 MQ 都是能容忍短暂的乱序，所以推荐使用普通顺序模式。
顺序消费的实现
在MQ的模型中，顺序需要由3个阶段去保障：
- 消息被发送时保持顺序
- 消息被存储时保持和发送的顺序一致
- 消息被消费时保持和存储的顺序一致
消息被发送时保持顺序
使用严格顺序模式
严格顺序消息模式下，对于指定的一个 Topic，所有消息按照严格的先入先出（FIFO）的顺序进行发布和消费。因此只要保证消息同步发送（发完一条后再发下一条）即可保证消息发送时保持顺序。
使用普通顺序模式
普通顺序模式下，只有同一个队列的消息能保证有序。Producer 生产消息的时候会进行轮询(根据设定的负载均衡策略)来向同一主题的不同消息队列发送消息。那么如果此时有几个消息分别是同一个订单的创建、支付、发货，在轮询的策略下这 三个消息会被发送到不同队列 ，因为在不同的队列此时就无法使用 RocketMQ 带来的队列有序特性来保证消息有序性了。
因此使用普通顺序时，在同步发送的基础上，还需要将消息发送到相同的队列。
在RocketMQ中，通过MessageQueueSelector来实现队列的选择。通过对订单的唯一标识符取hash，将同一个订单的消息发送到相同的队列。
消息被消费时保持和存储的顺序一致
在分布式的情况下，即使消息队列有序的将消息发送给消费者，也可能因为网络等原因，导致消费者接收到的消息无序。如：按顺序发送消息a、b给消费者。虽然a先发送，但因为网络原因，消息a在网络中滞留一段时间，导致消费者收到的消息顺序为b、a。同时，若同一个队列的消息由不同消费者消费也可能出现以上情况。
消费者顺序消费消息的实现
基于以上分析，要保证消息顺序的被消费者消费，必须满足下列条件：
- 同一个订单的消息由同一个消费者消费
- 消费者消费完一条消息之后，才可以接着消费下一条
同一个消费者消费
类似于通过订单id的hash选择相同的队列，可以通过订单的hash选择同一个消费者同步消费（消费完一条后再拉取下一条，单线程消费），保证同一个订单的顺序消费
通过 consumer 内部用内存队列做排队，然后分发给底层不同的 worker 实现（实现复杂）
若消费者是多线程，此时在消费者内部建立内存队列。先将消息拉取到内存队列后，在分发给不同的线程
消息丢失
消息的可靠性需要由3个阶段去保障：
- 发送端消息可靠性
- 存储端消息可靠性
- 消费端消息可靠性
发送端消息可靠性
消息发送一般有以下几种方式：同步发送、异步发送以及单向发送，业务具体选择哪种方式进行消息发送，需要根据情况进行判断，下面具体介绍不同的发送方式实现的消息可靠性保证。
（1）同步发送
同步发送是指发送端在发送消息时，阻塞线程进行等待，直到服务器返回发送的结果。发送端如果需要保证消息的可靠性，防止消息发送失败，可以采用同步阻塞式的发送，然后同步检查Brocker返回的状态来判断消息是否持久化成功。如果发送超时或者失败，则会默认重试2次，RocketMQ选择至少传输成功一次的消息模型，但是因为网络传输是不可靠的，有可能发生重复投递。
（2）异步发送
异步发送是指发送端在发送消息时，传入回调接口实现类，调用该发送接口后不会阻塞，发送方法会立即返回，回调任务会在另一个线程中执行，消息发送结果会回传给相应的回调函数。具体的业务实现可以根据发送的结果信息来判断是否需要重试来保证消息的可靠性。
（3）单向发送
单向发送是指发送端发送完成之后，调用该发送接口后立刻返回，并不返回发送的结果，业务方无法根据发送的状态来判断消息是否发送成功，单向发送相对前两种发送方式来说是一种不可靠的消息发送方式，因此要保证消息发送的可靠性，不推荐采用这种方式来发送消息。
存储端消息可靠性
存储端的可靠性依靠持久化策略、备份（主从复制）保证
RocketMQ刷盘机制
同步刷盘
消息写入内存的 PageCache后，立刻通知刷盘线程刷盘，然后等待刷盘完成，刷盘线程执行完成后唤醒等待的线程，返回消息写成功的状态。这种方式可以保证数据绝对安全，但是吞吐量不大。
异步刷盘
消息写入到内存的 PageCache中，就立刻给客户端返回写操作成功，当 PageCache中的消息积累到一定的量时，触发一次写操作，或者定时等策略将 PageCache中的消息写入到磁盘中。这种方式吞吐量大，性能高，但是 PageCache中的数据可能丢失，不能保证数据绝对的安全。
消费端消息可靠性
（1）消费重试
消费者从RocketMQ拉取到消息之后，需要返回消费成功来表示业务方正常消费完成。因此只有返回CONSUME_SUCCESS才算消费完成，如果返回CONSUME_LATER则会按照不同的messageDelayLevel时间进行再次消费，时间分级从秒到小时，最长时间为2个小时后再次进行消费重试，如果消费满16次之后还是未能消费成功，则不再重试，会将消息发送到死信队列，从而保证消息存储的可靠性。
（2）死信队列
未能成功消费的消息，消息队列并不会立刻将消息丢弃，而是将消息发送到死信队列，其名称是在原队列名称前加%DLQ%，如果消息最终进入了死信队列，则可以通过RocketMQ提供的相关接口从死信队列获取到相应的消息，保证了消息消费的可靠性。
（3）消息回溯
回溯消费是指Consumer已经消费成功的消息，或者之前消费业务逻辑有问题，现在需要重新消费。要支持此功能，则Broker存储端在向Consumer消费端投递成功消息后，消息仍然需要保留。重新消费一般是按照时间维度，例如由于Consumer系统故障，恢复后需要重新消费1小时前的数据。RocketMQ Broker提供了一种机制，可以按照时间维度来回退消费进度，这样就可以保证只要发送成功的消息，只要消息没有过期，消息始终是可以消费到的。