Mq的幂等性问题分析和基本处理

Posted 2021-10-04 codeWjh

前景介绍：

在看这篇文章之前我先说说几个我在RocketMq使用中总结的问题，如有错误，请联系我删除谢谢。

• 消息的发送失败和消费失败大多数都是网络原因和服务器内存问题。

• 但是消费者消费消息失败除了以上原因外，还关系到我们在消费的方法中是否会产生异常，一旦产生异常，消费者就会给broker返回ack为0的应答，那么这个消息就会存在broker里面不会被剔除，并且会触发消息的重试机制，只到返回ack为1的应答。

  知道上面这个知识后我们来思考一个问题，消费者消费失败后触发重试机制，这样就产生了重复消费问题。因此，使用MQ时应该对一些关键消息进行幂等去重的处理。那，如何进行消息的幂等性处理呢？

​ 博主最近遇到了一个业务流程。上图：

其实有很多业务流程一样的，比如支付成功后发送消息更改 “某些数据“ 的状态,一旦不做幂等性处理就会出现实际已付款，但是c端显示已收货。。。

1.为什么会出现消息被重复消费？

在解决消息的幂等性之前我们需要知道为什么消息会被重复消费？

我们通常会认为，消息中间件是一个可靠的组件，这里所谓的可靠是指，只要我把消息成功投递到了消息中间件，消息就不会丢。即消息至少会被消费者成功消费一次，这是消息中间件最基本的特性之一。

然而这种可靠的特性会导致消息可能被多次的投递。

看几个可能出现的场景：

• 优惠券微服务 接受到这个消息 M 并完成消费逻辑之后，正想通知消息中间件“我已经消费成功了”的时候，程序就宕机了，那么对于消息中间件来说，这个消息并没有成功消费过，所以它还会继续投递。这时候对于优惠券微服务 来说，看起来就是这个消息明明消费成功了，但是消息中间件还在重复投递。这在 RockectMQ 的场景来看，就是同一个 messageId 的消息重复投递下来了。那这个消息不久被重复的消费了吗？
• 我们一般在用户调用支付接口成功后才会修改订单状态并发送消息给优惠券系统，但是做过微信支付的人都知道，当用户支付成功后，微信后台会调用我们项目暴露的接口，这个时候你们绝对能想到网络的波动，微信后台由于网络波动一直调我们的接口，那这个消息不就被重复的发送了吗？
• 还有一种可能：采用同步消息发送，当网络出现波动，虽然消息已经发送到MQ了，当时MQ并没有把成功的响应给生产者，导致消息发送失败，实际已经发送出去了，但是生产者为了保证消息真的被发出去，会再发送一次消息。那这个消息不就被重复的发送了吗？

总结：经过我们的分析，我们发现真正的项目开发不像我们自己做练习项目那样，而是一定要考虑消息的幂等性问题，特别是涉及到钱相关的问题。

2.如何解决消息幂等性问题？

说了这么多问题可能产生的原因，那真正的如何解决呢？

在我遇到幂等性问题之后，我查询了多方资料，有了一定的了解，这次的随笔我不用代码说明，只有口头语言，也可能代表我的猜测，下次有时间我会出一个代码demo测试，最近有点忙哈，哈哈。

• 引入redis实现幂等性问题，但是注意这种方式不是100%成功。

  • 实现思路

    • @Component
      @Slf4j
      @RocketMQMessageListener(
              topic = "order",
              consumerGroup = "order-group"
      )
      public class OrderConsumerListener implements RocketMQListener<MessageExt> {
      
          @Autowired
          private StringRedisTemplate redisTemplate;
      
          @Override
          @Transactional
          public void onMessage(MessageExt messageExt) {
             //"key"搞一个消息的唯一标识，根据自身业务系统来设计
              String msgId = redisTemplate.boundValueOps("key").get();
              if(!StringUtils.isEmpty(msgId)) {
                  return;
              } else {
                  redisTemplate.boundValueOps("key").set("1",10L, TimeUnit.MINUTES);
                  //处理消费消息的业务
                  
              }
          }
      }
      

    • 仔细看我们发现这是一个线程不安全的，不具备原子性，这个是个风险，还有就是只能防止10分钟，如果10分钟之内消费不成功，还是会出现幂等问题，而且还可能发生极端情况：redis宕机了怎么办。但是不能否认这种方式可以加长key的过期时间，因为rocketmq里面消息的重试是有次数和时间间隔的，我们可以手动配置，注意：这里我们保证的消息幂等，不保证消息一定要消费成功，默认是失败了16次之后就进入死信队列。

看了这么久是不是发现还没有很完美的解决办法，其实博主目前接手的项目就是用redis处理的，因为有赞这边的服务器都比较好，顶多接口请求超时才可能发生，所以给10分钟的时间缓冲。。。所以说这种事情确实要根据自身硬件和业务系统的需求来完成。。。。但是最近还是会有一些问题，tl给出的解决方案是在myql层结合业务处理（同一条记录已经修改了的不能重复改，已经有记录的不能插，就类似是这种意思）。

其实最完美的办法应该是在数据库业务层进行处理，即数据库乐观锁。

由于时间有限，我不做说明。

什么是数据库乐观锁和悲观锁可以看下面引用的链接进行学习，mybatis-plus里面实现了乐观锁，乐观锁一般是基于update操作的。

https://www.cnblogs.com/kyoner/p/11318979.html

那insert操作如何处理呢？

其实insert操作我们可以把消息的唯一主键当成要插入记录的主键，这样就会报主键异常，在数据库层面解决了问题，但是也会在重试16次之后进入死信队列。

最后，这次时间太赶，没有更好的说明解决方案，也没有给出代码测试，后面如果遇到类似的问题我会找时间完善这篇文章的。谢谢大家的阅读，一入java深似海，越学越菜~~~~。