• 背景

最近项目开发到比较关键的阶段,在消息中间件上出现了一些问题,因为是电商项目,我们使用了springboot搭建了两个关键的微服务——会员系统(member system)和订单系统(order system),下面分别简称msos
其中,os负责产生交易订单orderIndfos,orderInfos里面的每一条数据orderInfo都绑定了一个会员。因为订单是会员产生的,这些数据需要投递给会员模块ms去计算成长值和积分,这个计算过程较为复杂,所哟普不能跟订单同步完成,所以只能通过消息中间件来传递,会员系统通过订阅这些数据进行成长值和积分的计算。消息中间件是RocketMQ

  • 问题发现

ms和os通过RocketMQ进行通信的期间,发生了一个意料之中的问题。——消息重复消费问题。
有过类似经验的同学应该对这类问题很熟悉,而没有经历过的同学,或许会感到疑惑,RocketMQ已经设置为单个的订阅模式(并非广播,按理说生产者不可能重复产生相同的消息),为什么消费者还会重复消费到相同的信息呢?
起初我也很疑惑,去查看平台记录,发现生产者的消息入所想的一样,只有一个,这说明问题跟生产者无关。而发生重复消费的原因,竟然是这个消息被重复订阅了好几次!换言之,同一个消息被重复投递了几次!

  • 查询原因

查看资料,得到了这样的解释:
一般来说,消息系统,对于未确认的消息,采用按规则重新投递的方式进行处理。如下情况可能会导致消息重复被投递:

1、订阅方应用接收到消息,业务处理完成后应用出中间件不知道消息处理结果,会重新投递消息。
2、订阅方应用接收到消息,业务处理完成后网络出中间件收不到消息处理结果,会重新投递消息。
3、订阅方应用接收到消息,业务处理时间过长,消息中间件因消息超时未确认,会再次投递消息。
4、订阅方应用接收到消息,业务处理完成,消息中间件问题导致收不到消息处理结果,消息会重新投递。
5、订阅方应用接收到消息,业务处理完成,消息中间件收到了消息处理结果,但由于消息存储故障导致消息没能成功确认,消息会再次投递。

  • 简要分析

这说明,这个问题的出现是十分复杂的。
上述的情况可能会出现在任何项目里面,而一般来说,情况2出现的可能性比较大,因为生产环境一旦发包调试,都是众多微服务一起启动,注册和发现过程中,环境不太稳定,很容易出现网络障碍,在这种情况下,RocketMQ中间件会收不到消费方的回复,从而重新投递消息。这属于mq本身应答异常状况的机制。

  • 解决方案
    既然找到了问题出现的原因,那我们就可以根据项目具体情况给出解决方案。

方案一: 分布式锁
我们知道,在生产环境,一般都是多元jvm集群环境,mq系统投递重复消息时,很有可能导致数据库重复消费(有人说可以采用synchronize或者Lock的方式锁住,但实际上行不通,因为同步锁只能保证多线程问题,注意,集群是多进程问题,不能用此方法解决)。就好比新增操作,一个消息新增一条数据,那么重复消息就会导致数据库新增两条重复数据,这明显是不符合业务场景的。那么,我们平时解决多进程问题最常用的方式就是分布式锁(一般是zookeeper和redis搭建)。分布式锁特有的集群全局视野,能够辅助,消息重复的甄别。
如下:

分布式

方案二:数据库约束+java异常处理机制

这个方案很简单,首先需要针对数据库简历约束,不允许产生重复数据,然后再使用java的异常处理机制来规避重复消息。如下:

public class TestClient {

    public static void main(String[] args) {
        // 生成消息工具类
        Student studentMessage = getStudent();
        try {
            usermapper.insertStudentid(studentMessage);
        }catch(Exception e) {
            // 异常处理
            if(e.getCause() instanceof MySQLIntegrityConstraintViolationException)
                log.warn("学号{}已存在,重新生成",studentMessage.getStudentid);
            //发现异常,也就说明消息重复了,业务中断,返回,直接丢掉消息
			return ;
        }
		//继续消费
		//…………其他后续业务;
    }

}

上述的异常处理是重点,请务必要有这个“e.getCause() instanceof MySQLIntegrityConstraintViolationException”条件,因为这个只处理相同消息的异常,仅仅是遇到这种异常,才会return MESSAGE.FAILURE;,否则还是要抛出。

这样,就完美解决了重复消息问题。当然很多人说这样做不优雅,显得很土。于是我们采用了第三种方案。

方案三:利用数据库有条件的插入语句限制重复插入

这个方案是最简单的方案,我们都知道mysql的语句,无论是insert还是update,都是支持有条件的执行的,update我就不多说了,就说一下insert。

insert into person_table (uid,pname,age)
select 2,"李四",20 from dual
where not exists
(select id from person_table where uid=2 )

上述语句中“where not exists”后面跟随的就是本条语句的插入条件。这样一来,问题就好办了。

在这里插入图片描述
如上图所示,此方案利用了数据库特性替代了锁。既方便,又可靠。

方案四:查询消息系统验证消息是否重复

方案二三都属于较为简单的解决方案,上述方案不能适用于所有场景。所以大部分大厂(如阿里,腾讯等等)是不会采用这种头痛医头脚痛医脚的方案的。既然问题是处在mq中,那么解铃还须系铃人。我们也可以直接请针对mq出方案。
一般来说,mq在发完消息,甚至消费者再订阅消息时,都会有记录。我们可以在消费时,通过订阅的记录和消费的结果来判断,此消息是否重复订阅过,倘若重复订阅,则不再数据库中插入数据。
此方案,关键在于,消费这个行为发生之前,必须去查询mq的订阅记录,验证此消息是否曾经发出来过,情况较为复杂,而且对后续人工发消息补偿时,也要纳入考虑,所有单单解决消息重复这一点来说,显得有些成本过高,所以我们没有采用。在这里也不做详细讨论。

当然,除了上面几种方案,必然还存在别的方案,如乐观锁,hash锁等等。欢迎大家一起讨论。
本文完。