消息队列怎么避免重复消费

Posted 2023-04-27

参考技术A 消息中间件是无法保证消息重复消费，所以只能从业务上来保证消费不重复消费，在消费端保证接口的幂等性。有以下两种方案。

MQ消息队列的重复消费问题的通用解决办法以及幂等性的原理

详细介绍了MQ消息队列重复消费的原因，以及通过保证幂等性来避免重复消费带来的问题。

文章目录

• 1 至少一次
• 2 重复消费的原因
• 3 幂等性处理重复消费

1 至少一次

消息领域有一个对消息投递的QoS定义（Quality of Service，服务质量），分为：最多一次（At most once）、至少一次（At least once）、仅一次（ Exactly once）。

目前火热的几款MQ，比如RocketMQ、Kafka、RabbitMQ、ActiveMQ等，都是保证的至少一次(At least Once)，它指每个消息必须投递一次。既然是至少一次，那么他们就避免不了重复消费的问题，因此这个问题最终还是要靠业务代码来解决。

2 重复消费的原因

重复消费的的原因大概可以分为两个，一个是生产者发送消息的时候发送了重复的消息，另一个是消费者消费的时候消费了重复的消息。

生产者发送消息的时候，如果我们只管发送，而不需要等待Broker响应消息发送成功，那么发往Broker的消息是不会重复的，除非自己的业务出错了（当然这也是一个原因）。

但是为了保证消息的可靠性，我们不可能调用发送接口就完事儿了，必须还得等待Broker的响应，这时就有可能出现问题了，如果某次发送消息之后，Broker的响应由于网络波动一时没有收到，那么当这个响应超出时间之后，通常生产者会因为没有收到响应而认为这条消息没有发送成功，此时生产者又会重复发送一次，最终导致两条消息都发送成功了，消息队列有两条重复的消息，这样就导致了消息的重复。

消费者消费消息的时候，如果业务逻辑已经走完了，那么就需要进行commit提交offset，如果此时消费者挂了，那么被消费但是没有被提交的消息将被认为是没有消费成功的消息而被分发到其他消费者上，导致一条消息被重复消费。

以上两个原因都是可能导致消息重复消费的原因，我们也能知道，网络波动或者服务挂掉等原因是不可能被100%避免的，因此消息的重复消费也不可能被避免。另外，RocketMQ消费者Rebalance的时候，可能出现拉取了消息还没有被成功消费并提交的时候，该队列被分配给了其他消费者，导致多个消费者消费同样的消息。

RocketMQ、Kafka等消息队列的说明中已经明确表示，消息队列本身不能处理和避免重复消费的情况，这需要业务人员自己处理。

既然无法避免重复消费，并且消息队列也无法处理，那么从业务或者说代码的角度处理重复消息呢？关键点就是幂等性。

3 幂等性处理重复消费

重复消费会带来的问题呢？如果消费者的业务逻辑是执行一个查询的逻辑，那么无论消费几次，这对于业务来说是没有太大的影响的，但如果消费者的业务逻辑是向数据库中插入一条数据，那么重复消费将会导致至少插入两条相同的数据，自然导致了数据的异常。

那么幂等是什么意思呢？幂等（idempotent、idempotence）实际上是一个数学与计算机学概念，在编程中一个幂等操作的特点是其任意多次执行所产生的影响均与一次执行的影响相同，通俗点说就是：同样的参数或者数据去调用同一个接口，无论重复调用多少次，总能保证数据的正确性，不能出错，这就是接口的幂等性。这里“数据的正确性”和具体的业务相关，不同的业务，对于幂等性的定义是不一样的。

基于幂等性的要求，我们需要改造业务处理逻辑，使得在重复消息的情况下也不会影响最终的结果。怎么改造呢？这得结合具体的业务来考虑：

1. 如果一个业务是只读业务，或者是更新的业务，那么多次读取或者多次更新相同的数据基本上都没什么问题。
2. 如果业务需要插入数据，但是插入的数据中有唯一key能够区分（比如订单id，或者生产者生成的token），那么业务逻辑就可以变成在插入前先查询这个唯一key是否在数据库中，如果消费方的业务表不需要存储这个key，那么消费方也可以单独建立一张唯一key表，插入唯一key表和插入业务表的sql逻辑一定要都在一个事务中，在插入前判断这个唯一key是否在数据库的唯一key表中，如果存在说明此前已经成功消费了这条消息，不再消费，否则就是没有成功消费，继续消费。
3. 如果业务需要插入数据，但是插入的数据中没有唯一key能够区分，这种是无法完全避免因为多条重复的消息或者一条消息多次重复消费带来的问题。因此最好是让发送消息的同事向消息内容中添加一个唯一的字段，即使消费者方的业务不需要也没关系，因为这样就能的防止重复的消息和重复消费了。
4. 上面通过数据库的方式来防止重复消费的都属于“强校验”类型，会一定程度上影响数据库性能，通常涉及到金钱的都需要强校验，如果不需要强校验，那么使用Redis来代替也行，比如生产者将id或者token作为key存入Redis，消费者消费时先判断是否存在id或者token，如果存在则消费，消费完毕之后消除key。很多的业务都不需要强校验，比如获取登陆短信验证码的时候，多发送两条，或者发送失败都没关系，我们都遇到过。

上面的都是通用的办法，但处理重复消费的问题，始终要根据具体业务来考虑自己的解决办法，比如我们公司有很多监听一张表的binlog日志然后将操作同步到另一张表上的场景，这种情况下，某些消费者直接将原始表的id作为同步表的id，这样插入的时候，如果id重复肯定是插入不了的，天然的就保证了消息的幂等性。但有时候，同步的表是使用的自己的自增id，此时就需要在插入之前通过其他唯一的业务字段判断此数据是否已被消费过，如果被消费过，则不执行插入。

如有需要交流，或者文章有误，请直接留言。另外希望点赞、收藏、关注，我将不间断更新各种Java学习博客！