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,同时也为他的天才设计感到敬佩!!!

  • 发表于:
  • 原文链接http://kuaibao.qq.com/s/20180425G1PSAX00?refer=cp_1026
  • 腾讯「云+社区」是腾讯内容开放平台帐号(企鹅号)传播渠道之一,根据《腾讯内容开放平台服务协议》转载发布内容。
  • 如有侵权,请联系 yunjia_community@tencent.com 删除。

扫码关注云+社区

领取腾讯云代金券