分布式事务消息最终一致性方案（转移）

随着分布式服务体系结构的流行和普及，最初在单个应用程序中执行的多个逻辑操作现在被拆分为多个服务之间的远程调用. 尽管服务化为我们的系统带来了横向扩展的能力，但挑战仍然是分布式事务问题. 多个服务使用它们自己单独维护的. 他们彼此之间不在同一笔交易中. A的执行成功，但是B的执行失败，此时A的事务已经提交，无法回滚，最终将导致双方数据不一致的问题；尽管很早以前就有基于两阶段提交的XA分布式事务，但是由于资源的全局锁定，这种解决方案的性能要求极低. 因此，稍后逐步推导出用于诸如消息最终一致性和TCC之类的灵活事务的分布式事务解决方案. 本文主要分析基于消息性计划的最终一致性. 节目.

通用邮件处理流程.png

消息生产者发送消息MQ接收消息，将消息持久化，在存储中添加一条记录，并向使用方返回ACK到相应的使用方MQ推送消息，然后等待使用方返回ACK. 消息使用者如果确认消息在指定时间内成功返回，则MQ认为消息使用成功，并从存储器中删除消息，即执行步骤6；否则，执行步骤6. 如果MQ在指定时间内未收到ACK，则认为消息使用失败，并将尝试再次推送消息重复步骤4、5和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数据量，处理方超时. 发生超时异常后，调用方将直接回滚. 本地事务，这导致在调用方上没有数据，但是在处理器端上的数据已被写入，最终导致双方业务数据中的不一致. 为了确保双方数据的一致性，我们只能从其他地方找到新的突破.

由于传统的处理方法无法解决消息产生者成功进行本地事务处理与消息成功传输之间的一致性问题，因此诞生了事务消息，实现了本地事务的原子性和消息传递的成功. 消息生产者. 它保证了消息生产者成功进行本地事务处理和消息成功传输之间的最终一致性.

交易消息处理流程. png

交易消息和普通消息之间的区别在于消息产生链接. 生产者首先向MQ预发送一条消息（也称为发送半条消息）

MQ收到消息后，将首先将其保留，然后将要发送状态的新消息添加到存储中

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

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

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

MQ将根据结果删除或更新消息状态为可发送

如果消息状态已更新为可发送，MQ会将消息推送给使用者，随后的消息的消耗与普通消息相同

注意: 由于MQ通常保证消息可以成功传递，因此，如果服务未及时返回ACK结果，则可能导致MQ的重复消息传递问题. 因此，对于最终的消息一致性方案，消息使用者必须支持消息使用的幂等性，这不能导致同一消息的重复使用.

用于异常处理的异常方法

不存储消息，不执行业务操作

一致

没有

存储要发送的消息成功，但是ACK失败，导致服务未执行（可能是由于MQ处理超时，网络抖动等）

不一致

MQ确认业务运营结果，处理消息（删除消息）

成功存储了要发送的消息，ACK成功和业务执行（可能成功或可能不成功），但是MQ没有收到生产者的业务处理的最终结果

不一致

MQ确认业务操作结果并处理消息（根据业务处理结果，更新消息状态，如果业务执行成功，则传递消息，如果失败，则删除消息）

业务流程成功，结果被发送到MQ，但是MQ更新消息失败，导致消息状态仍然被发送

不一致

同上.

当前更主流的MQ，例如ActiveMQ，RabbitMQ，Kafka，RocketMQ等，仅RocketMQ支持事务消息. 根据作者的理解，在早期几年中，阿里在MQ的交易新闻方面的增长也是由于支付宝的业务需求所致. 因此，如果我们要严重依赖MQ事务消息来实现最终的一致性，那么在当前情况下，只能通过选择RocketMQ来解决技术选择问题. 上面我们还分析了事务消息的异常，即MQ存储了要发送的消息，但是MQ无法感知上游处理的最终结果. 对于RocketMQ，其解决方案非常简单，即其内部实现将具有计划任务，以训练要发送的消息的状态，然后向生产者发送检查请求，生产者必须实现检查侦听器，侦听器内容通常是检查相应的本地事务是否成功（通常查询），如果成功，MQ会将消息设置为可发送，否则删除该消息.

问: 如果预发送消息失败，是否没有执行业务？

答案: 是的，对于基于消息最终一致性的解决方案，通常强烈依赖此步骤. 如果不能保证这一步，最终将不可能实现最终的一致性.

问: 为什么需要为两条已发布的消息添加消息预发送机制并添加重试机制？如果服务失败且传输失败，为什么不使用重试机制？

回答: 如果业务执行成功，则发送一条消息. 如果尚未发送消息，则业务系统将关闭. 重新启动系统后，没有记录是否曾经发送过消息，这将引起业务. 执行成功，消息未发出.

如果消费者消费失败，是否需要回退生产者？

答案: 对于此处的交易消息，由于消费者消费失败，生产者将不会回滚. 事务消息的应用寻求高可用性和最终的一致性. 如果消息使用失败，则MQ将负责. 推送消息分布式事务消息，直到使用成功. 因此，事务消息是针对生产方的，而对于消费者方，则是通过MQ重试机制来完成消费者方的一致性.

如果消费者端由于业务异常而导致回滚，那么双方最终不会不能保证一致性吗？

答案: 最终的基于消息的一致性解决方案必须确保不会阻碍消费者的业务运营. 它仅允许系统故障和业务故障，例如在您的业务上抛出NPE. 导致消费者执行交易的问题失败，很难实现一致性.

由于并非所有MQ都支持事务性消息，因此，如果我们不选择RocketMQ作为系统的MQ，是否可以实现消息的最终一致性？答案是肯定的.

基于本地messages.png的最终一致性

基于本地消息的最终一致性方案的核心方法是在执行业务操作时将消息数据记录到DB，并且消息数据和业务数据的记录必须在同一事务中完成. 这是程序前提的核心保证. 完成后记录了消息数据之后，我们可以稍后去DB训练要通过定时任务发送的消息，然后将消息传递给MQ. 在此过程中，可能会出现消息传递失败的可能性. 此时，重试机制用于确保更新消息状态或清除消息，直到成功接收到MQ ACK确认为止. 如果后续消息使用失败，则取决于MQ本身是否重试完成，并最终实现双方系统数据的最终一致性. 尽管基于本地消息服务的方案可以实现消息的最终一致性，但是它具有严重的缺点. 当每个业务系统使用此方案时，有必要在相应的业务库中创建消息表来存储消息. . 针对此问题，我们可以单独提取此函数，并使其成为消息服务以统一的方式进行处理，从而得出我们将在下面讨论的方案.

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

独立消息服务的最终一致性与本地消息服务的最终一致性之间的最大区别是，将消息的存储分别放入RPC服务中. 该过程实际上是预发送模拟交易消息的消息的过程. 如果预发送消息失败，那么将不执行生产者的业务，因此对于生产者的业务，它强烈依赖于消息服务. 幸运的是，独立的消息传递服务支持水平扩展，因此，只要将多个单元部署为HA集群模式，就可以保证其可靠性. 在消息服务中，还有一个单独的定时任务，该任务定期轮询处于长时间发送状态的消息，并通过检查补偿机制确认与该消息相对应的业务是否成功. 如果相应的业务处理成功，则将消息修改为可发送，然后将其传递给MQ；如果业务流程失败，请更新或删除相应的消息. 因此，在使用此解决方案时，消息生产者还必须为消息服务实现检查服务以确认消息. 对于消息的消耗，解决方案与上述处理相同，全部通过MQ自己的重传机制来确保消息被消耗.

本文来自电脑杂谈，转载请注明本文网址：
https://www.pc-fly.com/a/jisuanjixue/article-256176-1.html