MQ消息最终一致性解决方案

随着分布式服务架构的流行与普及，原来在单体应用中执行的多个逻辑操做，如今被拆分红了多个服务之间的远程调用。虽然服务化为咱们的系统带来了水平伸缩的能力，然而随之而来挑战就是分布式事务问题，多个服务之间使用本身单独维护的数据库，它们彼此之间不在同一个事务中，假如A执行成功了，B执行却失败了，而A的事务此时已经提交，没法回滚，那么最终就会致使两边数据不一致性的问题；尽管很早以前就有基于两阶段提交的XA分布式事务，可是这类方案由于须要资源的全局锁定，致使性能极差；所以后面就逐渐衍生出了消息最终一致性、TCC等柔性事务的分布式事务方案，本文主要分析的是基于消息的最终一致性方案。

普通消息的处理流程

1. 消息生成者发送消息
2. MQ收到消息，将消息进行持久化，在存储中新增一条记录
3. 返回ACK给生产者
4. MQ push 消息给对应的消费者，而后等待消费者返回ACK
5. 若是消息消费者在指定时间内成功返回ack，那么MQ认为消息消费成功，在存储中删除消息，即执行第6步；若是MQ在指定时间内没有收到ACK，则认为消息消费失败，会尝试从新push消息,重复执行四、五、6步骤
6. MQ删除消息

普通消息处理存在的一致性问题

咱们以订单建立为例，订单系统先建立订单(本地事务)，再发送消息给下游处理；若是订单建立成功，然而消息没有发送出去，那么下游全部系统都没法感知到这个事件，会出现脏数据；

public void processOrder() {
    // 订单处理(业务操做) 
    orderService.process();
    // 发送订单处理成功消息(发送消息) 
    sendBizMsg ();
}

复制代码

若是先发送订单消息，再建立订单；那么就有可能消息发送成功，可是在订单建立的时候却失败了，此时下游系统却认为这个订单已经建立，也会出现脏数据。

public void processOrder() {
   // 发送订单处理成功消息(发送消息) 
    sendBizMsg ();
    // 订单处理(业务操做) 
    orderService.process();
}

复制代码

一个错误的想法

此时可能有同窗会想，咱们能否将消息发送和业务处理放在同一个本地事务中来进行处理，若是业务消息发送失败，那么本地事务就回滚，这样是否是就能解决消息发送的一致性问题呢?

@Transactionnal
public void processOrder() {
    try{
        // 订单处理(业务操做) 
        orderService.process(); 
        // 发送订单处理成功消息(发送消息) 
        sendBizMsg ();
    }catch(Exception e){
         事务回滚;   
    }
}

复制代码

消息发送的异常状况分析

可能的状况	一致性
订单处理成功，而后忽然宕机，事务未提交，消息没有发送出去	一致
订单处理成功，因为网络缘由或者MQ宕机，消息没有发送出去，事务回滚	一致
订单处理成功，消息发送成功，可是MQ因为其余缘由，致使消息存储失败，事务回滚	一致
订单处理成功，消息存储成功，可是MQ处理超时，从而ACK确认失败，致使发送方本地事务回滚	不一致

从上面的状况分析，咱们能够看到，使用普通的处理方式，不管如何，都没法保证业务处理与消息发送两边的一致性，其根本的缘由就在于：远程调用，结果最终可能为成功、失败、超时；而对于超时的状况，处理方最终的结果多是成功，也多是失败，调用方是没法知晓的。 笔者就曾经在项目中出现相似的状况，调用方先在本地写数据，而后发起RPC服务调用，可是处理方因为DB数据量比较大，致使处理超时，调用方在出现超时异常后，直接回滚本地事务，从而致使调用方这边没数据，而处理方那边数据却已经写入了，最终致使两边业务数据的不一致。为了保证两边数据的一致性，咱们只能从其余地方寻找新的突破口。

事务消息

因为传统的处理方式没法解决消息生成者本地事务处理成功与消息发送成功二者的一致性问题，所以事务消息就诞生了，它实现了消息生成者本地事务与消息发送的原子性，保证了消息生成者本地事务处理成功与消息发送成功的最终一致性问题。

事务消息处理的流程

1. 事务消息与普通消息的区别就在于消息生产环节，生产者首先预发送一条消息到MQ(这也被称为发送half消息)

2. MQ接受到消息后，先进行持久化，则存储中会新增一条状态为待发送的消息

3. 而后返回ACK给消息生产者，此时MQ不会触发消息推送事件

4. 生产者预发送消息成功后，执行本地事务

5. 执行本地事务，执行完成后，发送执行结果给MQ

6. MQ会根据结果删除或者更新消息状态为可发送

7. 若是消息状态更新为可发送，则MQ会push消息给消费者，后面消息的消费和普通消息是同样的

注意点：因为MQ一般都会保证消息可以投递成功，所以，若是业务没有及时返回ACK结果，那么就有可能形成MQ的重复消息投递问题。所以，对于消息最终一致性的方案，消息的消费者必需要对消息的消费支持幂等，不能形成同一条消息的重复消费的状况。

事务消息异常状况分析

异常状况	一致性	处理异常方法
消息未存储，业务操做未执行	一致	无
存储待发送消息成功，可是ACK失败，致使业务未执行(多是MQ处理超时、网络抖动等缘由)	不一致	MQ确认业务操做结果，处理消息(删除消息)
存储待发送消息成功，ACK成功，业务执行(可能成功也可能失败)，可是MQ没有收到生产者业务处理的最终结果	不一致	MQ确认业务操做结果，处理消息(根据就业务处理结果，更新消息状态，若是业务执行成功，则投递消息，失败则删除消息)
业务处理成功，而且发送结果给MQ，可是MQ更新消息失败，致使消息状态依旧为待发送	不一致	同上

支持事务消息的MQ

如今目前较为主流的MQ，好比ActiveMQ、RabbitMQ、Kafka、RocketMQ等，只有RocketMQ支持事务消息。据笔者了解，早年阿里对MQ增长事务消息也是由于支付宝那边由于业务上的需求而产生的。所以，若是咱们但愿强依赖一个MQ的事务消息来作到消息最终一致性的话，在目前的状况下，技术选型上只能去选择RocketMQ来解决。上面咱们也分析了事务消息所存在的异常状况，即MQ存储了待发送的消息，可是MQ没法感知到上游处理的最终结果。对于RocketMQ而言，它的解决方案很是的简单，就是其内部实现会有一个定时任务，去轮训状态为待发送的消息，而后给producer发送check请求，而producer必须实现一个check监听器，监听器的内容一般就是去检查与之对应的本地事务是否成功(通常就是查询DB)，若是成功了，则MQ会将消息设置为可发送，不然就删除消息。

常见的问题

1. 问：若是预发送消息失败，是否是业务就不执行了？

   答：是的，对于基于消息最终一致性的方案，通常都会强依赖这步，若是这个步骤没法获得保证，那么最终也 就不可能作到最终一致性了。

2. 问：为何要增长一个消息预发送机制，增长两次发布出去消息的重试机制，为何不在业务成功以后，发送失败的话使用一次重试机制？

   答：若是业务执行成功，再去发消息，此时若是还没来得及发消息，业务系统就已经宕机了，系统重启后，根本没有记录以前是否发送过消息，这样就会致使业务执行成功，消息最终没发出去的状况。

3. 若是consumer消费失败，是否须要producer作回滚呢？

   答：这里的事务消息，producer不会由于consumer消费失败而作回滚，采用事务消息的应用，其所追求的是高可用和最终一致性，消息消费失败的话，MQ本身会负责重推消息，直到消费成功。所以，事务消息是针对生产端而言的，而消费端，消费端的一致性是经过MQ的重试机制来完成的。

4. 若是consumer端由于业务异常而致使回滚，那么岂不是两边最终没法保证一致性?

   答：基于消息的最终一致性方案必须保证消费端在业务上的操做没障碍，它只容许系统异常的失败，不容许业务上的失败，好比在你业务上抛出个NPE之类的问题，致使你消费端执行事务失败，那就很难作到一致了。

因为并不是全部的MQ都支持事务消息，假如咱们不选择RocketMQ来做为系统的MQ，是否可以作到消息的最终一致性呢？答案是能够的。

基于本地消息的最终一致性

基于本地消息的最终一致性方案的最核心作法就是在执行业务操做的时候，记录一条消息数据到DB，而且消息数据的记录与业务数据的记录必须在同一个事务内完成，这是该方案的前提核心保障。在记录完成后消息数据后，后面咱们就能够经过一个定时任务到DB中去轮训状态为待发送的消息，而后将消息投递给MQ。这个过程当中可能存在消息投递失败的可能，此时就依靠重试机制来保证，直到成功收到MQ的ACK确认以后，再将消息状态更新或者消息清除；然后面消息的消费失败的话，则依赖MQ自己的重试来完成，其最后作到两边系统数据的最终一致性。基于本地消息服务的方案虽然能够作到消息的最终一致性，可是它有一个比较严重的弊端，每一个业务系统在使用该方案时，都须要在对应的业务库建立一张消息表来存储消息。针对这个问题，咱们能够将该功能单独提取出来，作成一个消息服务来统一处理，于是就衍生出了咱们下面将要讨论的方案。

独立消息服务的最终一致性

独立消息服务最终一致性与本地消息服务最终一致性最大的差别就在于将消息的存储单独地作成了一个RPC的服务，这个过程其实就是模拟了事务消息的消息预发送过程，若是预发送消息失败，那么生产者业务就不会去执行，所以对于生产者的业务而言，它是强依赖于该消息服务的。不过好在独立消息服务支持水平扩容，所以只要部署多台，作成HA的集群模式，就可以保证其可靠性。在消息服务中，还有一个单独地定时任务，它会按期轮训长时间处于待发送状态的消息，经过一个check补偿机制来确认该消息对应的业务是否成功，若是对应的业务处理成功，则将消息修改成可发送，而后将其投递给MQ；若是业务处理失败，则将对应的消息更新或者删除便可。所以在使用该方案时，消息生产者必须同时实现一个check服务，来供消息服务作消息的确认。对于消息的消费，该方案与上面的处理是同样，都是经过MQ自身的重发机制来保证消息被消费。

总结：上游事务提交以后，在基于MQ的场景下就不考虑回滚了。失败的多是因为网络、服务宕机所致使，文章中提到说业务上执行是无障碍的。若是下游服务长时间没有恢复，那么就应该设置告警，在这里有几种机制来解决一些牛皮癣类型的问题，假如上游消息始终发送失败（这种可能性基本不存在除非代码是假的）这种状况咱们能够设置报警机制好比发生异常时能够打印日志，发送短信，发送邮件，将异常订单保存到数据库，这些措施能够同时用于下游一些异常订单，同时也能够在发生异常的时候新建一个异常Topic的消息提示，让人工来介入数据订正。

这篇文章若是对你有帮助 点个关注吧