在上一篇MQ系列文章中，小编为大家讲解了如何用MQ解决并发问题。其实，MQ的功能不仅止于此，它还能用来解决分布式事务。本期，小编就为大家讲解MQ高阶知识：如何基于MQ提供分布式事务解决方案。

1 适用场景

当应用/服务/数据库间产生分布式事务时，若对数据时效性要求不是太严格的情况下，可以通过MQ来实现事务最终一致性。

注：1. 有强事务需求的可以使用TCC，使用方法可参考文章“处理分布式事务效率低下？苍穹TCC助你一臂之力”；

2. 关于MQ的配置方法，可参考文章“MQ入门：跑通第一个单机案例”。

2 需求背景

功能性需求：单据提交后，自动执行审核操作。

非功能性需求：

1）单据提交失败，审核也失败：数据一致；

2）单据提交成功，审核失败，无法回滚提交：数据不一致×。

3 解决方案

3.1 工作原理与核心思路

基于MQ的分布式事务解决方案的工作原理如下图所示：

工作原理示意

该解决方案的核心思路如下：

• 在微服务架构下，分布式事务框架确保了多应用实例分布于不同的节点之上的事务的最终一致性。

• 使用可靠消息队列MQ，保证上下游应用数据操作的最终一致性，即把本地业务逻辑和消息发送记录放在一个事务中，保证本地操作和消息记录的状态一致：要么都成功，要么都失败。下游应用向消息系统订阅该消息，收到消息后执行相应操作。

• 从本质上讲，使用MQ来解决分布式事务的原理是将分布式事务转换为两个本地事务，然后依靠下游业务的重试机制达到最终一致性。

3.2 关键节点说明

对于消息发送者：

“1.BeginTransaction”和“5.EndTransaction”：消息发送记录需要发生在外层业务逻辑的事务保护中；

“3. 本地事务表”和“4. 发送MQ消息”：将事务的xid写入消息及消息发送表t_dtx_trans中，若表、索引等不存在，则会动态创建表和索引（理论上，该表每个库都会有）。

消息发送者相关节点代码运行示意

“线程捡漏”：若发送消息到MQ的过程失败，但消息发送记录成功保存到t_dtx_trans，则可以通过线程捡漏扫描该表，并通过state字段过滤识别需要重新投递的消息。

线程捡漏节点代码运行示意

对于消息接受者：

“Ⅰ. 消费消息”：消费者线程获取消息进行消费，消息状态由“ready”变为“unacknowledged”。

参数说明：

1）ready：消息已到达MQ，等待消费；

2）unacknowledged：消息正在被消费，未确认；

3）total：ready+ unacknowledged。

“Ⅱ. 判断源事务是否提交”：根据消息中记录的xid查询t_dtx_trans表，即可判断事务是否已提交。若已提交，则正常消费，并使unacknowledged-1；若事务还未提交，则会将该消息重新投递至队列头部，等待消费。当重试次数达到max次数后，则会从队列中discard（丢弃）消息。

判断源事务是否提交代码运行示意

“Ⅲ. 执行消息方法”：默认自动确认，但建议手动确认。相关代码说明如下：

1) acker.ack(messgeID)；//消息确认

2) acker.deny(messgeID)；//消息重试

3) acker.discard(messgeID)；//消息丢弃

3.3 设计关键点

该解决方案的设计关键点在于保证可靠事件投递和避免重复消费，通过以下措施来实现：

(1) 消息队列发送事务性支持：确保每个服务的业务操作和消息发布事件的原子性。

注：如果把一个事务看作一个程序，它要么完整地被执行，要么完全不执行。这种特性就叫原子性。

(2) 利用消息确认机制ACK实现消息队列的可靠性投递，确保事件至少传递一次。

注：如果在处理消息的过程中，消费者的服务器在处理消息的时候出现异常，则该条消息可能没有完成消息消费，导致数据丢失。为了确保数据不会丢失，RabbitMQ支持消息确定-ACK。

ACK消息确认机制是指消费者从RabbitMQ收到消息并处理完成后，反馈给RabbitMQ，RabbitMQ收到反馈后才将此消息从队列中删除。

（3）下游业务执行的顺序及幂等性控制。

（4）消息可靠持久化。

3.4 缺陷

（1）若采用publishInDbTranscation，在消费者侧，应注意步骤Ⅱ“判断源事务是否提交”。当发送者事务提交失败时，unacknowledged的消息会直接压回队列的头部。

当unack的消息个数＞=消费者个数时，消费者一致循环处理unacknowledged的消息，导致后面ready的消息阻塞，只能等到重试次数达到上限，自动删除消息后，方可消费后面的ready消息。

（2）缺少可视化的事务维护界面，若重试次数达到需要手工干预的时候，业务人员不方便排查及处理。

4 操作步骤与实现效果

1.打开单据设计器，编辑提交操作，添加插件kd.demo.sci.opplugign.AutoAuditOP。

添加插件操作示意

2. 点击“提交”，程序运行如下：

提交操作代码运行示意

3. 手动构造网络异常，捕获异常进行重试，程序运行如下：

捕获异常进行重试代码运行示意

4. 处理异常，模拟好网络环境后，逻辑正常，进行消息确认。

消息确认代码运行示意

5. 最终结果：已审核。

最终实现效果示意

5 附件

1.kd.demo.sci.opplugign.AutoAuditOP操作插件代码如下：

package kd.demo.sci.opplugign;
import kd.bos.dataentity.entity.DynamicObject;
import kd.bos.entity.plugin.AbstractOperationServicePlugIn;
import kd.bos.entity.plugin.args.EndOperationTransactionArgs;
import kd.bos.mq.MQFactory;
import kd.bos.mq.MessagePublisher;
public class AutoAuditOP extends AbstractOperationServicePlugIn {
   @Override
   public void endOperationTransaction(EndOperationTransactionArgs e) {
      MessagePublisher mp = MQFactory.get().createSimplePublisher("kded_tc", "erkai_queue");
      try {
         for(DynamicObject data:e.getDataEntities()) {
            //p1:该操作所在的分库
            //p2:自定义数据
            mp.publishInDbTranscation("secd",data.get("id") );
         }
      } finally {
         mp.close();
      }
   }
}

2. kd.demo.sci.consumer.DemoConsumer消费者类相关代码如下：

package kd.demo.sci.consumer;
import java.io.IOException;
import java.util.List;
import kd.bos.dataentity.OperateOption;
import kd.bos.entity.operate.result.OperateErrorInfo;
import kd.bos.entity.operate.result.OperationResult;
import kd.bos.exception.KDBizException;
import kd.bos.logging.Log;
import kd.bos.logging.LogFactory;
import kd.bos.mq.MessageAcker;
import kd.bos.mq.MessageConsumer;
import kd.bos.orm.query.QCP;
import kd.bos.orm.query.QFilter;
import kd.bos.servicehelper.QueryServiceHelper;
import kd.bos.servicehelper.operation.OperationServiceHelper;
public class DemoConsumer implements MessageConsumer {
   Log log = LogFactory.getLog(getClass());
   @Override
   public void onMessage(Object id, String messageId, boolean resend, MessageAcker acker) {
      log.info("自定义DemoConsumer开始消费");
      try {
         OperationResult result = OperationServiceHelper.executeOperate("audit", "kded_testt", new Object[] {id}, OperateOption.create());
         ListallErrorInfo = result.getAllErrorInfo();
         if(allErrorInfo.size()<=0) {
            acker.ack(messageId);
            return;
         }
         QFilter q1 = new QFilter("id",QCP.equals,id);
         QFilter q2 = new QFilter("billstatus",QCP.equals,"C");
         //如果这条数据已经审核了则废弃消息不再进行重试
         boolean discard = QueryServiceHelper.exists("kded_testt", new QFilter[] {q1,q2});
         if (discard){
            acker.discard(messageId);//废弃
            log.info(id+"已经审核，此消息丢弃");
         } else{
            acker.deny(messageId);//告诉mq重发这条消息
            log.error(result.getMessage());
         }
      } catch (Exception e) {
         acker.deny(messageId);//告诉mq重发这条消息
         // 记录异常原因，并写业务日志
         log.error(e);
      }
   }
}

以上就是本期的全部内容啦，后续小编还会带来MQ服务调用模型的详细介绍

#往期推荐#

# MQ入门：跑通第一个单机案例

# MQ进阶：如何用MQ解决并发问题

# 想要玩转MQ？那就必须懂MQ服务调用模型！

更多精彩内容，“码”上了解！↓

MQ高阶：基于MQ的分布式事务解决方案

MQ高阶：基于MQ的分布式事务解决方案

恭喜您！