RocketMQ系列之消息重试＋重复消费

前言

上节我们介绍了RMQ的几大模块以及每个模块的作用，前面也介绍了RMQ的订阅消费案例，今天我们来看一个RMQ的消息重试机制和重复消费的问题，了解这2点有助于我们更好更合理的处理消息消费的异常情况。

为什么会出现消息重试？

因为RMQ的消息都是通过网络传输的，通过网络传输就难免会受网络环境的影响，各种可能存在的情况，可能导致生产者Producer发送消息失败，也可能导致消费者Consumer消费消息失败，因此RMQ的消息重试机制就显得比较重要了，这也是RMQ的一大优势所在，显然消息重试机制分2种。

生产者Producer端重试

生产端发送消息失败就是指，Producer向MQ发送消息的时候没有发送成功，导致的原因可能有网络传输失败等，下面我们就来看看生产端是怎么处理消息发送失败的：

配置生产者的重试次数：消息重试发送的次数限制

配置生产者发送消息时的超时等待：在指定时间内如果消息没有成功发送到MQ就尝试重新发送

Producer端代码如下：

消费者Consumer端重试

消费者消费消息失败就是指，Consumer从MQ取到消息进行消费的过程中，由于某些原因导致消费失败（网络原因，消息逻辑处理异常，消费者直接宕机等等），下面我们就来看下消费端是怎么处理消息失败的：

设置消费最大重试次数：默认是16，当配置的值大于16的时候，第16次之后就会每次重试时间间隔2小时，当配置的值小于等于16时，重试的间隔时间如下图：

从Broker的启动日志也能发现这一点：

通过返回重试状态码：Consumer提供了2个状态码

CONSUME_SUCCESS：消息消费成功状态，如果返回该状态，那么对应的这条消息就会从RMQ上被消费完成并移出MQ

RECONSUME_LATER：消息消费重试状态，如果返回该状态，消费者会在间隔时间内再次尝试消费该消息，每尝试一次之后，该消息对应的reconsumeTimes的值+1，默认第一次失败时为0，不算重试次数

消息消费重试实践

消费者消费失败又分多种情况，下面我们将一次次来实践一遍看看效果：

第一种情况：消费者处理消息逻辑时异常

Consumer端代码如下：

在这里，为了验证之前说的Group组的概念，我开了2个Consumer端，属于同一个组，我们其实可以看到消息是被负载的分给2个Consumer的，我们看下面运行结果：

我们再看下该消息对应的重试次数参数变化：

第二种情况：当同一个组中的消费者Consumer宕机之后，MQ会将消息转发给剩下的其他Consumer处理，包括失败重试的消息也一样会转到其他Consumer中被处理（是不是突然对这个Group组有着莫名的好感）

我们看下面这个例子：我先启动了2个Consumer，这个重试的消息落在了c1的身上，当重试一次之后，我把c1宕机了，我们来看看c2发生了什么，第2条消息被转发到了c2上，而且重试次数也是在之前c1的基础上操作的；

第三种情况：当同一个组下的某个Consumer处理的消息超时的时候，MQ消息就会不断尝试处理这条消息，直到发送成功为止（这个是RMQ内部自己做的重试机制），这种情况是不会转发给另一个Consumer处理的：生产者生产一条消息，被c2处理了，c2睡眠60秒，在这60秒内，消息都是一直在c2上进行重试（隐式实现），直到我把c2宕机，你会发现，消息才会被c1处理（上述第二种情况）：

讲到这里，其实我们的消息重试就差不多讲完了，但是有一点一定要注意

注意：消费端的消息重试机制一定要在集群消费模式下才有效，广播消费模式下，RMQ是不会进行重试机制的，广播模式下，消息只消费一次，不管你有没有成功！！！

消息重复消费问题

之前我们讲过当我们先启动生产者生产消息，后启动消费者消费消息时，当多个消费者就有可能消费到同一条消息，就像2个人去领任务，第一个人先领取了任务1，但是还在处理，任务还没完成，第二个人过来时，也看到了任务1，就也领取了任务1，然后就造成2个人处理了同一个任务，我们可以看下面示例：c1和c2同时处理了第一条消息，很明显这是不合理的

对于上面的问题，我们就需要相应的处理策略，我总结觉得可以从下面2个方面入手

保证消费端处理消息的业务逻辑保持幂等性

如何保证幂等呢，我们主要从以下几个手段考虑：

幂等性可以自己业务逻辑实现，例如不管逻辑代码执行多少次，只要是同一个编号处理，得到的结果都是一样的，例如更新订单状态，只要是同一个订单号，就算重复消费，执行了多个update，最终数据库还是一样的结果；

如果不是update这种操作呢，例如insert一条订单下单成功记录，那么此时我们可以通过设置数据库表某个字段唯一约束，例如订单号，来解决处理结果的幂等；

如果insert的数据不能设置唯一约束呢，那么我们还可以启动一个脚本，定时扫描数据库表，发现如果是同样的数据被生成出来，可以删掉一条，以此来保证重复消费带来的数据重复；

总之，不管你用什么办法，就是假如消息被重复消费了，那么我们一定要想办法来保证执行结果的幂等。

保证每条消息都有唯一标识，且每条消息只会被处理一次

既然上面是假如消息被重复消费了，那么当然还有一个办法就是防止消息被重复消费主要有下面2个手段：

利用一张日志表来记录已经处理成功的消息的ID，如果新到的消息ID已经在日志表中，那么就不再处理这条消息

我们可以给每条消息自定义一个状态字段，当生产消息时默认为未消费状态，当获取到消息时，标为正在消息状态，当消费完时标为已消费状态（这一步可以不做，因为当一个消息被成功消费完时，其实他也就不在RMQ中了，其他消费者也不会获取到这条消息）。然后每次消费者消费消息时，都先对消息这个状态值进行判断，如果是正在消费或者已消费就不做处理，直接获取下一条

OK，以上就是我们今天所讲的消息重试和重复消费问题，希望看完，能对您有所帮助，便于你能更好的使用RMQ，同时也为他的天才设计感到敬佩！！！