在众多关于MQ的面试八股文中有这么一道题“如何保证MQ消息消费的幂等性”。为什么需要保证幂等性呢是因为消息会重复消费。为什么消息会重复消费明明已经消费了为什么消息会被再次被消费呢不同的MQ产生的原因可能不一样本文就以RocketMQ为例来扒一扒RocketMQ中会导致消息重复消息的原因最终你会发现其实消息重复消费算是RocketMQ无奈的“bug”。消息发送异常时重复发送首先我们来瞅瞅RocketMQ发送消息和消费消息的基本原理。如图简单说一下上图中的概念Broker就是RocketMQ的服务端如上图就有两个服务实例Topic就是一类消息集合的名字Queue就是Topic的对应的队列消息都存在Queue上每个Topic都会有自己的几个Queue所以整个消息发送和消费过程大致如下生产者在发送消息之前根据负载均衡策略(默认是轮询)选择一个Queue然后跟这个Queue所在的机器建立连接把消息发送到这个Queue上消费者只要消费这个Queue那么就能消费到消息在正常情况下生产者的确是按照这个方式来发送消息的但是当出现了异常时这种异常包括消息发送超时、响应超时等等RocketMQ为了保证消息成功发送会进行消息发送的重试操作默认情况下会最多会重试两次重试操作比较简单就是选择另一台机器的Queue来发送。虽然重试操作可以很大程度保证消息能够发送成功但是同时也会带来消息重复发送的问题。举个例子假设生产者向A机器发送消息发生了异常响应超时了但是就一定代表消息没发成功么不一定有可能会出现服务端的确接收到并处理了消息但是由于网络波动等等导致生产者接收不到服务端响应的情况此时消息处理成功了但是生成者还是以为发生了异常此时如果发生重试操作那么势必会导致消息被发送了两次甚至更多次导致服务端存了多条相同的消息那么就一定会导致消费者重复消费消息。消费消息抛出异常在RocketMQ的并发消费消息的模式下需要用户实现MessageListenerConcurrently接口来处理消息当消费者获取到消息之后会调用MessageListenerConcurrently的实现传入需要消费的消息集合msgs这里提到的msgs很重要如上代码当消息消费出现异常的时候status就会为null后面就会将status设置成为RECONSUME_LATER。RECONSUME_LATER翻译成功中文就是稍后重新消费的意思所以从这可以看出一旦抛出异常那么消息之后就可以被重复消息。到这其实可能有小伙伴觉得消息消费失败重新消费很正常保证消息尽可能消费成功。对这句话不错的确可以在一定程度上保证消费异常的消息可以消费成功。但是坑不在这而是前面提到的消费时传入的整个集合中的消息都需要被重新消费。具体的原因我们接着往下看当消息处理之后不论是成功还是异常都需要对结果进行处理代码如下当处理结果为RECONSUME_LATER的时候异常会设置为RECONSUME_LATER此时ackIndex会设置成-1后面循环遍历的时候就会遍历到所有这次消费的消息然后调用sendMessageBack方法sendMessageBack方式是用来实现消息重新消费的逻辑这里就不展开说了。所以一旦被消费的一批消息中出现一个消费异常的情况那么就会导致整批消息被重新消费从而会导致在出现异常之前的成功处理的消息都会被重复消费非常坑。不过好在消费时传入的消息集合中的消息数量是可以设置的并且默认就是1也就说默认情况下那个集合中就一条消息所以默认情况下不会出现消费成功的消息被重复消费的情况。所以这个参数不要轻易设置一旦设置大了就可能导致消息被重新消费。除了并发消费消息的模式以外RocketMQ还支持顺序消费消息的模式也会造成重复消费逻辑其实差不多但是在实现消息重新消费的逻辑不一样。消费者提交offset失败首先来讲一讲什么是offset。前面说过消息在发送的时候需要指定发送到消息最后会被放到Queue中其实真正的消息不是在Queue中Queue存的是每个消息的位置但是你可以理解为Queue存的是消息。而消息在Queue中是有序号的这个序号就被称为offset从0开始单调递增1。比如说如上图消息1的offset就是0消息2的offset就是1依次类推。这个offset的一个作用就是用来管理消费者的消费进度。当消费者在成功消费消息之后需要将所消费的消息的offset提交给RocketMQ服务端告诉RocketMQ这个Queue的消息我已经消费到了这个位置了。提交offset的代码就在上述第二节提到的处理结果的后面这样有一个好处那么一旦消费者重启了或者其它啥的要从这个Queue拉取消息的时候此时他只需要问问RocketMQ服务端上次这个Queue消息消费到哪个位置了之后消费者只需要从这个位置开始消费消息就行了这样就解决了接着消费的问题。但是RocketMQ在设计的时候当消费完消息的时候并不是同步告诉RocketMQ服务端offset而是定时发送。如图当消费者消费完消息的时候会将offset保存到内存中的一个Map数据结构中所以上面截图的那段代码其实是更新内存中的offset而在消费者启动的时候会开启一个定时任务默认是5s一次会通过网络请求将内存中的每个Queue的消费进度offset发送给RocketMQ服务端。由于是定时任务所以就可能出现服务器一旦宕机导致最新消费的offset没有成功告诉RocketMQ服务端的情况此时消费进度offset就丢了那么消费者重启的时候只能从RocketMQ中获取到上一次提交的offset从这里开始消费而不是最新的offset出现明明消费到了第8个消息RocketMQ却告诉他只消费到了第5个消息的情况此时必然会导致消息又出现重复消费的情况。服务端持久化offset失败上一节说到消费者会有一个每隔5s钟的定时任务将每个队列的消费进度offset提交到RocketMQ服务端当RocketMQ服务端接收到提交请求之后会将这个消费进度offset保存到内存中同时为了保证RocketMQ服务端重启消费进度不会丢失也会开启一个定时任务默认也是5s一次将内存中的消费进度持久化到磁盘文件中所以整个消费进度offset的数据流转过程如下当RocketMQ服务端重启之后会从磁盘中读取文件的数据加载到内存中。跟消费者产生的问题一样一旦RocketMQ发生宕机那么offset就有可能丢失5s钟的数据RocketMQ服务端一旦重启消费者从RocketMQ服务端获取到的消息消费进度就比实际消费的进度低同样也会导致消息重复消费。主从同步offset失败在RocketMQ的高可用模式中有一种名叫主从同步的模式当主节点挂了之后从节点可以手动升级为主节点对外提供访问保证高可用。在主从同步模式下从节点默认每隔10s会向主节点发送请求同步一些元数据这些元数据就包括消费进度当从节点获取到主节点的消费进度之后会将主节点的消费进度设置到自己的内存中同时也会持久化到磁盘。所以整个消费进度offset的数据的流转过程就会变成如下同样由于也是定时任务那么一旦主节点挂了从节点就会丢10s钟的消费进度此时如果从节点升级为主节点对外提供访问就会出现跟上面提到的一样的情况消费者从这个新的主节点中拿到的消费进度比实际的低自然而然就会重复消费消息。所以总的来说在消费进度数据流转的过程中只要某个环节出现了问题都有很有可能会导致消息重复消费。重平衡先来讲一讲什么是重平衡其实重平衡很好理解我说一下你就明白了。前面说到消费者是从队列中获取消息的在RocketMQ中有个消费者组的概念一个消费者组中可以有多个消费者不同消费者组之间消费消息是互不干扰的所以前面提到的消费者其实都在消费组下在同一个消费者组中消息消费有两种模式集群消费模式广播消费模式由于RocketMQ默认是集群消费模式并且绝大多数业务场景都是使用集群消费模式所以这里就不讨论广播消费模式了感兴趣的同学可以看看RocketMQ消息短暂而又精彩的一生 这篇文章。集群消费模式是指同一条消息只能被这个消费者组消费一次这就叫集群消费。并且前面提到提交消费进度给RocketMQ服务端的情况只会集群消费模式下才会有在广播消费模式不会提给到RocketMQ服务端仅仅持久化到本地磁盘同时前面说的消费者提交消费进度真正提交的是消费者组对于这个Queue的消费进度而不是指具体的某个消费者对于Queue消费进度。虽然说这里将前面提到的一些含义更深一步但是并不妨碍前面的理解。集群消费的实现就是将队列按照一定的算法分配给消费者默认是按照平均分配的。如图所示假设某个topic有4个Queue有个消费者组订阅了这个topic这个消费者组有两个消费者1和消费者2此时每个消费者就可以被分配两个队列这样就能保证消息正常情况下只会被消费一次。如果只有一个消费者那么这个消费者就会消费所有队列很好理解。接着后面又启动了一个消费者3此时为了保证刚上线的消费者3能够消费消息就要进行重平衡操作重新分配每个消费者消费的队列。在重平衡之后就可能会出现下面这种情况如上图原本被消费者2消费的Queue4被分配给消费者3此时消费者3就能消费到消息了这就是重平衡。除了新增消费者会导致重平衡之外消费者数量减少队列的数量增加或者减少都会触发重平衡。在了解了重平衡概念之后接下来分析一下为什么重平衡会导致消息的重复消费。假设在进行重平衡时还未重平衡完之前消费者2此时还是会按照上面第二节提到的消费消息的逻辑来消费Queue4的消息当消费者2已经重平衡完成了发现Queue4自己已经不能消费了那么此时就会把这个Queue4设置为dropped就是丢弃的意思但是由于重平衡进行时消费者2仍然在消费Queue4的消息但是当消费完之后发现队列被设置成dropped那么此时被消费者2消费消息的offset就不会被提交原因如下代码这段代码前面已经出现过一旦dropped被设置成true这个if条件就通不过消费进度就不会被提交。成功消费消息了但是却不提交消费进度这就非常坑了。。于是当消费者3开始消费Queue4的消息的时候他就会问问RocketMQ服务端我消费者3所在的消费者组对于Queue4这个队列消费到哪了我接着消费就行了。此时由于没有提交消费进度RocketMQ服务端告诉消费者3的消费进度就会比实际的低这就造成了消息重复消费的情况。清理长时间消费的消息在RocketMQ中有这么一个机制会定时清理长时间正在消费的消息。如图假设有5条消息现在正在被消费者处理这5条消息会被存在一个集合中并且是按照offset的大小排序消息1的offset最小消息5的offset最大。RocketMQ消费者启动时会开启一个默认15分钟执行一次的定时任务这个定时任务会去检查正在处理的消息的第一条消息也就是图中的消息1一旦发现消息1已经处理了超过15分钟了那么此时就会将消息1从集合中移除之后会隔一定时间再次消费消息1。这也会有坑虽然消息1从集合中被移除了但是消息1并没有消失仍然被消费者继续处理但是消息1隔一定时间就会再次被消费就会出现消息1被重复消费的情况。这就是清理长时间消费的消息导致重复消费的原因。但此时又会引出一个新的疑问为什么要移除这个处理超过15分钟的消息呢这就又跟前面提到的消费进度提交有关前面说过消息被消费完成之后会提交消费进度提交的消费进度实际会有两种情况第一种就是某个线程消费了所有的消息当把所有的消息都消费完成之后就会把消息从集合中全部移除此时提交的消费进度offset就是图中消息5的offset1加1的操作是为了保证如果发生重启那么消费者下次消费的起始位置就是消息5后面的消息保证消息5不被重复消费第二种情况就不太一样了假设现在有两个线程来处理这5条消息线程1处理前2条线程2处理后3条如图现在线程1出现了长时间处理消息的情况。此时线程2处理完消息之后移除后面三条消息准备提交offset的时候发现集合中还有元素就是线程1正在处理的前两条消息此时线程2提交的offset并不是消息5对应的offset而是消息1的offset代码如下这么做的主要原因就是保证消息1和消息2至少被消费一次。因为一旦提交了消息5对应的offset如果消费者重启了下次消费就会接着从消息5的后面开始消费而对于消息1和消息2来说并不知道有没有被消费成功就有可能出现消息丢失的情况。所以一旦集合中最前面的消息长时间处理那么就会导致后面被消费的消息进度无法提交那么重启之后就会导致大量消息被重复消费。为了解决这个问题RocketMQ引入了定时清理的机制定时清理长时间消费的消息这样消费进度就可以提交了。最后总得来说RocketMQ中还是存在很多种导致消息重读消费的情况并且官方也说了只是在大多数情况下消息不会重复所以如果你的业务场景中需要保证消息不能重复消费那么就需要根据业务场景合理的设计幂等技术方案。