[Java知识库]分布式事务解决方案和代码落地

在学习Spring cloud alibaba Seata之前，我们先来了解一下分布式事务的常用解决方案和代码实现，看完出去面试再遇到分布式事务的问题，战无不胜。文章包括了2PC，3PC刚性事务；TCC，本地消息表，可靠性消息，双写对账，最大努力通知，sage事务等柔性事务，并且大多数都有代码参考。

1.什么是分布式事务

完成某一个业务功能可能需要横跨多个服务，操作多个数据库,这个时候，我们本地事务就无法保证每个服务对数据库的操作是同时成功或者同时失败，这个时候就需要我们了解分布式事务。

1.1 分布式事务场景

1.跨库事务 ?

? ? ? ? ?跨库事务指的是，一个应用某个功能需要操作多个库。

2.分库分表

通常一个库数据量比较大或者预期未来的数据量比较大，都会进行水平拆分，也就是分库分表。如下图，将数据库B拆分成了2个库：

对于分库分表的情况，一般开发人员都会使用一些数据库中间件来降低sql操作的复杂性。如，对于sql：insert into user(id,name) values (1,"张三"),(2,"李四")。这条sql是操作单库的语法，单库情况下，可以保证事务的一致性。

但是由于现在进行了分库分表，开发人员希望将1号记录插入分库1，2号记录插入分库2。所以数据库中间件要将其改写为2条sql，分别插入两个不同的分库，此时要保证两个库要不都成功，要不都失败，因此基本上所有的数据库中间件都面临着分布式事务的问题。

3.服务化

在我们微服务中，服务与服务之间通过Rpc远程调用，假如ServiceA 里面调用了B然后调用C，调用B成功了，C失败了，但是服务B缺无法回滚了。或者ServiceA成功调用BC，但是准备提交本地事务的时候宕机了，这个时候B和C已经提交了。这个时候就会造成数据的不一致。

1.2.CAP原则和BASE理论

之前的文章已经介绍了很多，有兴趣可以看一下我之前的博客。

SpringCloud Alibaba(四) Nacos服务端本地启动和源码浅析(AP架构),Distro协议介绍,CAP原则介绍_Dr.劳的博客-CSDN博客_springcloud本地启动

2.分布式事务解决方案（刚性事务--强一致）

2.1 两阶段提交2PC

2.1.1 DTP（Distributed Transaction Processing）模型

如果一个系统操作多个数据库，肯定是有跨多个库的分布式事务的一个问题，在很多年之前全世界，老美早就已经发现这个问题了，很早以前就定义了一整套的解决方案来处理分布式事务的问题 有个叫做X/Open的组织定义了分布式事务的模型（DTP）。

这里面有几个角色，就是

AP（Application，应用程序）

TM（Transaction Manager，事务管理器），

RM（Resource Manager，资源管理器），

CRM（Communication Resource Manager，通信资源管理器）

然后这里定义了一个很重要的概念，就是全局事务，这个玩意儿说白了就是一个横跨多个数据库的事务，就是一个事务里，涉及了多个数据库的操作，然后要保证多个数据库中，任何一个操作失败了，其他所有库的操作全部回滚，这就是所谓的分布式事务。

2.1.2 XA规范

上面这套东西就是所谓的X/Open组织搞的一个分布式事务的模型，那么XA是啥呢？说白了，就是定义好的那个TM与RM之间的接口规范，就是管理分布式事务的那个组件跟各个数据库之间通信的一个接口。

比如管理分布式事务的组件，TM就会根据XA定义的接口规范，刷刷刷跟各个数据库通信和交互，告诉大家说，各位数据库同学一起来回滚一下，或者是一起来提交个事务把，大概这个意思

这个XA仅仅是个规范，具体的实现是数据库产商来提供的，比如说MySQL就会提供XA规范的接口函数和类库实现，等等

2.1.3 两阶段提交协议

X/Open组织定义的一套分布式事务的模型，还是比较虚的，还没办法落地，而且XA接口规范也是一个比较务虚的一个东西，光靠我说的这些东西还是没法落地的 基本上来说，你搞明白了XA也就明白了2PC了，2PC说白了就是基于XA规范搞的一套分布式事务的理论，也可以叫做一套规范，或者是协议。

Two-Phase-Commitment-Protocol，两阶段提交协议 2PC，其实就是基于XA规范，来让分布式事务可以落地，定义了很多实现分布式事务过程中的一些细节。

2.1.4 生活案例

??

2.1.5 二阶段提交代码落地

• 1.生成user库和account库的连接对象
• 2.生成全局事务id和分支事务id（后面会讲解这两个Id的作用）并且绑定，user库一个分支事务Id，account库一个分支数据id。
• 3.TM向user库（RM）发起请求依次执行xa.start，需要执行的sql，xa.end。
• 4.TM向account库（RM）发起请求依次执行xa.start，需要执行的sql，xa.end。
• 5.TM询问user库和account库准备好了吗？
• 6.两个RM都回答Ok,提交事务
• 7.如果有一个异常或者回复不Ok,回滚事务。

    public static void main(String[] args) throws SQLException {
        //true表示打印XA语句,，用于调试
        boolean logXaCommands = true;
        // 获得资源管理器操作接口实例 RM1
        Connection conn1 = DriverManager.getConnection
                ("jdbc:mysql://localhost:3306/db_user", "root", "root");
        XAConnection xaConn1 = new MysqlXAConnection(
                (com.mysql.jdbc.Connection) conn1, logXaCommands);
        XAResource rm1 = xaConn1.getXAResource();
        // 获得资源管理器操作接口实例 RM2
        Connection conn2 = DriverManager.getConnection
                ("jdbc:mysql://localhost:3306/db_account", "root", "root");
        XAConnection xaConn2 = new MysqlXAConnection(
                (com.mysql.jdbc.Connection) conn2, logXaCommands);
        XAResource rm2 = xaConn2.getXAResource();
        // AP请求TM执行一个分布式事务，TM生成全局事务id
        byte[] gtrid = "g12345".getBytes();
        int formatId = 1;
        try {
            // ==============分别执行RM1和RM2上的事务分支====================
            // TM生成rm1上的事务分支id
            byte[] bqual1 = "b00001".getBytes();
            Xid xid1 = new MysqlXid(gtrid, bqual1, formatId);
            // 执行rm1上的事务分支
            rm1.start(xid1, XAResource.TMNOFLAGS);//One of TMNOFLAGS, TMJOIN, or TMRESUME.
            PreparedStatement ps1 = conn1.prepareStatement(
                    "INSERT into user(name) VALUES ('test')");
            ps1.execute();
            rm1.end(xid1, XAResource.TMSUCCESS);
            // TM生成rm2上的事务分支id
            byte[] bqual2 = "b00002".getBytes();
            Xid xid2 = new MysqlXid(gtrid, bqual2, formatId);
            // 执行rm2上的事务分支
            rm2.start(xid2, XAResource.TMNOFLAGS);
            PreparedStatement ps2 = conn2.prepareStatement(
                    "INSERT into account(user_id,money) VALUES (1,10000000)");
            ps2.execute();
            rm2.end(xid2, XAResource.TMSUCCESS);

            // ===================两阶段提交================================
            // phase1：询问所有的RM 准备提交事务分支
            int rm1_prepare = rm1.prepare(xid1);
            int rm2_prepare = rm2.prepare(xid2);

            // phase2：提交所有事务分支
            boolean onePhase = false;
            //TM判断有2个事务分支，所以不能优化为一阶段提交
            if (rm1_prepare == XAResource.XA_OK
                    && rm2_prepare == XAResource.XA_OK) {
                //所有事务分支都prepare成功，提交所有事务分支
                rm1.commit(xid1, onePhase);
                rm2.commit(xid2, onePhase);
            } else {
                //如果有事务分支没有成功，则回滚
                rm1.rollback(xid1);
                rm2.rollback(xid2);
            }
        } catch (XAException e) {
            // 如果出现异常，也要进行回滚
            e.printStackTrace();
        }
    }

3.1.3 TCC代码演示

这里我用的开源框架是byteTcc作为例子,我们只需要看一下如何将一个接口拆分为try，comfirm，cancel，即使你用seata也是一样的流程，框架会自动帮你调用每个微服务对应的try，comfirm，cancel，无需我们自己手动去操作。并且他们会帮你记录事务日志，如果没用框架，哪一个微服务执行了try，confirm，cancel都需要保存在我们事务日志表，保存日志和业务操作在同一个事务提交。

代码逻辑：

我们模拟UserClient2转账给UserClient1，调用的UserClient1#transfer，两个client都标注了@Compensable，证明这是一个需要开启分布式事务的方法，这个注解里面有一个接口，这个接口里面的方法就对应Try，例如client1是transfer ，client2是decreaseAmount? ?。

UserClient1:

try：#transfer? ?假设B转账给A的10元，用户A的冻结资金字段+10

confirm：#confirmTransfer? 用户A的真实资金字段+10，冻结资金-10，就是恢复0，表示这次转账完成

cancel：#cancelTransfer? 用户A的冻结资金字段-10

UserClient2:（这里我们冻结字段-10也可以，不必纠结）

try：#decreaseAmount? ?用户B自己的冻结资金字段加上+10，并且真实资金-10

confirm：#decreaseAmountConfirm? 用户B冻结资金 -10，就是恢复0，表示这次转账完成

cancel：#decreaseAmountCancel 用户B的冻结资金 -10，真实资金+10

首先UserClient1#transfer，这个方法里面会通过Feign去扣除UserClient2的钱，并且UserClient2的扣钱方法也是Try方法，然后Client1的本地事务为自己加钱的Try方法，事务提交，框架会帮我们自动调用client1和client2的comfirm方法，如果有异常框架会自动帮我们调用他们的cancel方法。记住切勿自己把异常捕获。

从这里看出，TCC分布式事务是我们自己手动写冻结资源，提交，补偿接口，会将一个接口拆分为3个，加大了系统复杂度，但是不会锁资源，并且可控度很高。

? UserClient1:

public interface ITransferService {

	public void transfer(String sourceAcctId, String targetAcctId, double amount);

}
----------------------------------------------------------------------------------------
@Compensable(interfaceClass = ITransferService.class, simplified = true)
@RestController
public class SimplifiedController implements ITransferService {
	
	@Autowired
	private TransferDao transferDao;
	@Autowired
	private IAccountService acctServiceFeign;
	
 /**
*ITransferService 实现了transfer这个方法，默认byteTcc会认为他是Try方法
*/
	@ResponseBody
	@RequestMapping(value = "/simplified/transfer", method = RequestMethod.POST)
	@Transactional
	public void transfer(@RequestParam String sourceAcctId, @RequestParam String targetAcctId, @RequestParam double amount) {
		this.acctServiceFeign.decreaseAmount(sourceAcctId, amount);
		this.increaseAmount(targetAcctId, amount);
	}

	private void increaseAmount(String acctId, double amount) {
		System.out.println("update tb_account_two set frozen = frozen + #{amount} where acct_id = #{acctId}");
		int value = this.transferDao.increaseAmount(acctId, amount);
		if (value != 1) {
			throw new IllegalStateException("ERROR!");
		}
		System.out.printf("exec increase: acct= %s, amount= %7.2f%n", acctId, amount);
	}

	@CompensableConfirm
	@Transactional
	public void confirmTransfer(String sourceAcctId, String targetAcctId, double amount) {
		System.out.println("update tb_account_two set amount = amount + #{amount}, frozen = frozen - #{amount} where acct_id = #{acctId}");
		int value = this.transferDao.confirmIncrease(targetAcctId, amount);
		if (value != 1) {
			throw new IllegalStateException("ERROR!");
		}
		System.out.printf("done increase: acct= %s, amount= %7.2f%n", targetAcctId, amount);
	}

	@CompensableCancel
	@Transactional
	public void cancelTransfer(String sourceAcctId, String targetAcctId, double amount) {
		System.out.println("update tb_account_two set frozen = frozen - #{amount} where acct_id = #{acctId}");
		int value = this.transferDao.cancelIncrease(targetAcctId, amount);
		if (value != 1) {
			throw new IllegalStateException("ERROR!");
		}
		System.out.printf("exec decrease: acct= %s, amount= %7.2f%n", targetAcctId, amount);
	}

}

UserClient2:

public interface IAccountService {
	public void decreaseAmount(String accountId, double amount);
}
-----------------------------------------------------------------------------------------
@Compensable(interfaceClass = IAccountService.class, confirmableKey = "accountServiceConfirm", cancellableKey = "accountServiceCancel")
@RestController
public class AccountController implements IAccountService {
	
	@Autowired
	private JdbcTemplate jdbcTemplate;
	 /**
*IAccountService 实现了decreaseAmount这个方法，默认byteTcc会认为他是Try方法
*/
	@ResponseBody
	@RequestMapping(value = "/decrease", method = RequestMethod.POST)
	@Transactional
	public void decreaseAmount(@RequestParam("acctId") String acctId, @RequestParam("amount") double amount) {
		int value = this.jdbcTemplate.update(
				"update tb_account_one set amount = amount - ?, frozen = frozen + ? where acct_id = ?", amount, amount, acctId);
		if (value != 1) {
			throw new IllegalStateException("ERROR!");
		}
		System.out.printf("exec decrease: acct= %s, amount= %7.2f%n", acctId, amount);

		// throw new IllegalStateException("error");
	}

	@CompensableConfirm
	@Transactional
	public void decreaseAmountComfirm(String acctId, double amount) {
		int value = this.jdbcTemplate.update("update tb_account_one set frozen = frozen - ? where acct_id = ?", amount, acctId);
		if (value != 1) {
			throw new IllegalStateException("ERROR!");
		}
		System.out.printf("done decrease: acct= %s, amount= %7.2f%n", acctId, amount);
	}

	@CompensableCancel
	@Transactional
	public void decreaseAmountCancel(String acctId, double amount) {
		int value = this.jdbcTemplate.update(
				"update tb_account_one set amount = amount + ?, frozen = frozen - ? where acct_id = ?", amount, amount, acctId);
		if (value != 1) {
			throw new IllegalStateException("ERROR!");
		}
		System.out.printf("undo decrease: acct= %s, amount= %7.2f%n", acctId, amount);
	}

	
}

3.1.4 TCC事务会产生的问题一：幂等性

幂等性问题会发生在我们confirm和cancel阶段，try也会但是很少，当我们所有服务调用try接口成功的时候，我们会调用对应服务的confirm或者cancel，这个时候由于网络原因超时了，会有一个retry的重试操作，网络超时代表我们并不知道confirm或者cancel的执行结，假如你进行重试，如果没保证幂等性就会产生数据的错误，所以我们必须要保证幂等性。

解决方案：其实tcc事务执行，会有一个贯穿整个全局事务的全局事务id并且每一个分支事务会有一个分支事务id，我们每个微服务本地需要有一张分支本地事务日志表，里面有的字段，全局事务id，分支事务id，分支事务执行状态（1.try执行成功，2.confirm执行成功，3cancel执行成功），这样我们在重试的时候，首先用分支事务id来作为锁的key，然后去查询本地事务表，我是否执行过这一步操作，如果执行过则不执行，这样就可以保证幂等性。所以我们业务操作每一步操作的时候都需要在本地事务表记录当前分支事务的状态，和业务代码一起提交事务，这样可以回溯分支事务是否完成。

例子：

开始全局事务：生成全局事务id=123

分支事务=1，全局事务id=123.? ? 扣商品库存

分支事务=2，全局事务id=123.? ? 生成订单

分支事务=3，全局事务id=123.? ? 赠送积分

分支事务=4，全局事务id=123.? ? 扣除余额

结束全局事务：异步删除对应的事务id

3.1.5 TCC事务会产生的问题二：事务悬挂

还是网络原因产生的问题，假设我们设置了请求的超时时间为3秒，当我们对服务B执行try操作的时候，产生了超时，这个时候我们会调用B对应的cancel接口，就是这么骚，我们的try还没执行，然后执行了cancel，这个时候就有可能产生脏数据。这个时候服务B执行try的线程有空回来了，然后执行了try操作，那就完蛋了，我这个try操作永远都悬挂在这里了，芭比Q，因为我已经执行了cancel。

解决方案：全局事务id，分支事务id，分支本地事务日志表，在我们执行try操作之前，加锁，然后去本地事务表找一下当前有没有执行过cancel操作，有就不执行。

3.1.6?TCC事务会产生的问题三：空回滚

拿下面的下单流程为例，假设步骤1执行try成功了，然后步骤2失败了，这个时候框架会帮我们自动去调用步骤1,2,3,4的cancel方法，然而2,3,4的try方法都还没执行，证明都还没开始预留资源，你他么就把我回滚了，我try还没扣余额，你cancel反倒还给我加余额，赢两次。这就是空回滚问题。

解决方案：全局事务id，分支事务id，分支本地事务日志表，在我们执行cancel操作之前，加锁，然后去本地事务表找一下当前有没有执行过try操作，有就执行，没有就不执行，并且记录本地事务日志表状态。

1.扣商品库存

2.生成订单

3.赠送积分

4.扣除余额

3.1.7?TCC事务会产生的问题三：cancel或者confirm失败

有很多好奇的小朋友就要问，假设我所有try都成功了，或者有一些失败了，需要cancel和confirm的时候因为网络问题失败了，肿么办。不要慌，本地会有个定时任务，定时去本地事务表日志扫描还未完成的事务，假设这个事务所有try都成功，有一部分confirm失败了，定时任务会不断去帮你执行confirm操作，反之cancel。这个时候仍然需要依赖我们全局事务id，分支事务id，本地事务日志表。

如果最后多次重试都失败必须人工补偿，要预警。

3.1.8?TCC事务会产生的问题四：宕机恢复

有很多好奇的小朋友就要问，假设我执行事务的中途宕机了怎么办，不要慌。项目启动的时候去去本地事务日志表扫描还没执行完的事务，然后去询问是不是所有分支事务的try都执行完了啊，如果是，定时任务会不断去帮你执行confirm操作，反之cancel。这个时候仍然需要依赖我们全局事务id，分支事务id，本地事务日志表。

如果最后多次重试都失败必须人工补偿，要预警。

3.1.9 TCC等框架简单原理

本地事务日志表，全局事务id，分支事务id对于TCC分布式事务非常重要，但是一般框架都帮我们给实现了这个功能，包括幂等性，空回滚，失败重试，宕机启动，包括整个try之后confirm或者cancel的自动调用，假如我们自己实现就需要解决这种问题。

一般这种框架原理是动态代理数据源，重写了commit等方法，在分布式事务开启的时候生成全局事务id，通过feign调用的时候，通过feign的拦截器，把全局事务id在请求头带过去，假如调用的服务也是在分布式事务中的话，就生成分支事务id，然后在commit本地事务的时候把当前分支事务的状态也记录下来了，因为在一个库所以能保证原子性。每个微服务本地事务commit之后try阶段结束，框架会自动帮我们调用和我们服务调用链相关联的confirm或者cancel。

3.2.0?TCC分布式事务总结

优点：并发高，不锁资源（2PC会锁数据库资源），本地事务提交即可，一般适用于资金等不能出错的场景。

缺点：复杂，需要一个业务接口需要拆成3个，业务侵入性非常之大，老系统改得你怀疑人生，如果自己实现还得记录本地事务日志，重试，一般团队没实力不会使用这个，因为太骚了。

3.2 本地消息表

1. A系统在自己本地一个事务里操作同时，插入一条数据到消息表（同一个数据库原子性）
2. 接着A系统将这个消息发送到MQ中去
3. B系统接收到消息之后，在一个事务里，往自己本地消息表里（幂等性）插入一条数据，同时执行其他的业务操作，如果这个消息已经被处理过了，那么此时这个事务会回滚，这样保证不会重复处理消息
4. B系统执行成功之后，就会更新自己本地消息表的状态以及A系统消息表的状态
5. 如果B系统处理失败了，那么就不会更新消息表状态，那么此时A系统会定时扫描自己的消息表，如果有没处理的消息，会再次发送到MQ中去，让B再次处理
6. 这个方案保证了最终一致性，哪怕B事务失败了，但是A会不断重发消息，直到B那边成功为止，达到最终一致
7. 我们发送出去的消息，必须得带上业务相关联的数据，例如新增用户，发送增加积分mq消息，我们的mq消息的消息体得带上新增这个用户的id，以便以后回溯查找问题。
8. 简单理解：消息本地化，定时任务定时发送消息中间件。

3.2.1 本地消息表事例

已用户注册赠送积分为例

1.用户服务插入用户数据

2.在同一个事务中，插入一条赠送积分的数据到本地消息表，状态为未发送。

3.事务提交

4.定时任务定时去本地消息表，找状态是未发送的数据，发送mq，然后状态改为已发送。

5.mq持久化消息。

6.积分服务监听对应的topic，消费消息的时候一定要注意幂等性，因为步骤4有可能重复发送消息，有可能网络问题，导致消息重复发送出去，mq在没有ack的情况下也会一直重复发送，我们的定时任务没有把消息的状态修改，然后定时任务会重复发送。我们要做的是消费消息的时候先用消息id插入本地消息表，保证幂等性，然后再去处理我们的业务。幂等性很多方法都可以保证，不一定需要记录消费的消息。例如消费的时候加上分布式锁，锁里面去查询业务是否完成，如果完成了下一次消息过来，就不消费。消费完消息记得ack，不然mq也会重复投递。

7.（可选）.积分服务消费完，告诉用户服务的本地消息表，把对应的消息改为已完成。

8.（可选）定时任务定时去本地消息表，找状态是已发送状态的数据，重复投递。

9.（兜底）如果一个消息多次投递失败，或者多次消费失败就不再投递或者不再重复消费，这个时候我们就得做预警，人工处理。

10.（改进）我们可以本地事务提交之后立即发送mq，然后修改本地消息的状态，这样实时性会好一点，其他还是不变，定时任务用于兜底和重发。

11.（改进）我们可以新增多一个查询接口，让消费端可以来查询我们业务的状态。

3.2.2 本地消息表实现代码

1.UserClient新增用户，并且保存本地消息增加积分表，本地积分表的数据必须带上关联业务的id等，这里我们带上了用户id，因为后续我们出异常了有可能去用户服务查询这个用户是否真的注册了。

2.定时任务投递状态为待发送的消息给mq，并且修改状态为已发送

3.PointClient接受处理消息，先插入消息表，标识已经处理过了，并且这个表有唯一约束，然后，处理业务消息。

4.记得ack，不然mq会重复投递。

5.（可选）业务处理完我们也可以告诉调用方处理完，消息状态改为已完成。

UserClient 用户服务

@Component
public class UserClient {
    @Autowired
    private UserMapper userMapper;
    @Autowired
    private PointLogMapper pointLogMapper;
    //本地事务，可以优化本地事务提交之后立刻发送mq，实时性会好一点
    @Transactional
    public ApiResult save(User user){
        //保存用户
        userMapper.save(user);
        PointLog pointLog=new PointLog();
        pointLog.setUserId(user.getid);
        pointLog.setPoint(1);
        pointLog.setcreateTimeUtc(TimeUtils.getUTCTime());
        pointLog.setstate(0);
        //保存积分日志记录
        pointLogMapper.savePointLog(pointLog);
    }
     
    //本地消息表找未发送mq的消息，发送mq
    @Scheduled(cron = "0 1 0 * * ?")
    public pushMessage(){
        List<PointLog> pointLogList=pointLogMapper.findAllLogUnPush();
        for (PointLog pointLog:
        pointLogList) {
            boolean success=sendMq(JSON.toJsonString(pointLog));
            if(success){
                pointLog.setstate(1);
                pointLogMapper.update(pointLog);
            }
        }
    }
}

PointClient积分服务?

@Component
public class PointClient {
    @Autowired
    private PonitMessageMapper pointMessageMapper;

    @Override
    @Transactional
    public void onMessage(String message){
        PointMessage pointMessage=JSONObject.parseObject(jsonObject.toJSONString(), new TypeReference<PointMessage>(){});
        //插入消息表，唯一约束
        pointMessageMapper.insert(pointMessage);
        //处理逻辑
        pointServeice.addPoint(pointMessage);
        
        message.ack();
    }
}

3.2.3 本地消息表总结

优点：

简单，开发成本低，简单

缺点：

与业务场景绑定，高耦合，不可公用

本地消息表与业务数据表在同一个库，占用业务系统资源，量大可能会影响数据库性能

需要加入MQ中间件，其实不需要也可以，我们自己去调用HTTP这样也可以。

3.3 可靠消息最终一致性方案

这个方案其实就是我们本地消息表的一个变种，实现原理基本也一样。

我们需要一个可靠消息服务，来帮我们把本地消息表还有定时任务的功能给做了，事务开始我们只需要投递一个半消息（消费端不可消费的消息），可靠消息服务把消息持久化（可以看做本地消息表），然后告诉可靠消息服务，我这次业务本地事务是提交还是回滚，如果是提交，我就让这个半消息的状态改变，变为可消费，并且消费端如果消费失败我们可靠消息服务的定时任务会不断投递可消费的消息给消费端，消费端ack。假如可靠消息服务迟迟收不到我是应该把这条半消息提交或者回滚，他会自动去调用发送端的查询接口，让他告诉我应该把这个半消息状态改为可消费，还是应该删除。

例如新增用户，增加积分

1.我用户服务先投递一个增加积分（要带上新增用户的id）的半消息给我们的可靠消息服务，投失败了本地事务不执行，投递成功了积分服务消费端也看不见这条消息。

2.执行本地事务新增用户。

3.本地事务成功，告诉可靠消息服务，你可以把之前的半消息变为可消费了，这样积分服务就能消费这条消息，注意幂等性。

4.本地事务失败，告诉可靠消息服务，你帮我把刚刚的半消息删除吧，我失败了。

5.假设因为网络原因迟迟没有告诉可靠消息服务，可靠消息服务会带着那条消息，来问你的用户服务，我这个用户是否生成了，如果生成了你就回复我用户生成了，我把消息的状态改为可消费，否则删除这条消息。

6.消费端消费和本地消息表一模一样，要告诉可靠消息我消费完了这条消息，你不要再发给我了，也要保住幂等性。

而这个可靠消息服，就是Rocketmq的事务消息功能，他完成了我们上述的所有功能。

3.3.1 rocketmq事务消息

RocketMQ的设计中broker与producter端的双向通信能力，使得broker天生可以作为一个事务协调器存在，而RocketMQ本身提供的存储机制为事务消息提供了持久化能力，RocketMQ的高性能机制以及可靠消息机制则为事务消息在系统发生异常时依然能够保证达成事务的最终一致性。 ? ? 在RocketMQ 4.3后实现了完整的事务消息，实际上其实是对本地消息表的要给封装，将本地消息移动到了MQ内部，解决Producter端的消息发送与本地事务执行的原子性问题。

若Producer 本地事务执行失败则自动向MQServer发送rollback消息，MQ Server接收到rollback消息后 将删除”增加积分消息“ 。

MQ订阅方（积分服务）消费消息，消费成功则向MQ回应ack，否则将重复接收消息。这里ack默认自动回应，即程序执行正常则自动回应ack。

5、事务回查

如果执行Producer端本地事务过程中，执行端挂掉，或者超时，MQ Server将会不停的询问同组的其他?Producer 来获取事务执行状态，这个过程叫事务回查。MQ Server会根据事务回查结果来决定是否投递消息。

以上主干流程已由RocketMQ实现，对用户侧来说，用户需要分别实现本地事务执行以及本地事务回查方法，因此只需关注本地事务的执行状态即可。

3.3.2 rocketmq事务消息代码落地

同样以新增用户，增加积分为例子。

用户服务新增用户，而这里的积分服务消费消息，注意保证幂等性。

注意我们有一张事务表，来保存对应的事务id是否已经完成，用于回查，所以我们需要一个事务id贯穿整个事务，下面用的txNo。

AccountClient 用户服务

1.accountInfoController#accountCreateMessage

我们使用UUID生成了一个TxNo作为事务id，用户回查事务，然后调用service的发送新增用户消息。

@RestController
@Slf4j
public class AccountInfoController {
    @Autowired
    private AccountInfoService accountInfoService;

    @PostMapping(value = "/create")
    public String createAccount(@RequestParam("accountNo")String accountNo){
        //创建一个事务id，作为消息内容发到mq
        String tx_no = UUID.randomUUID().toString();
        AccountChangeEvent accountChangeEvent = new AccountChangeEvent(accountNo,tx_no);
        //发送消息
        accountInfoService.createAccountMessage(accountChangeEvent);
        return "创建用户";
    }
}

2.accountInfoService#createAccountMessage 这个方法里面发送了rocketmq事务消息到producer_group_txmsg，这个时候是一个半消息，消费端看不见。

@Service
@Slf4j
public class AccountInfoServiceImpl implements AccountInfoService {

    @Autowired
    AccountInfoDao accountInfoDao;

    @Autowired
    RocketMQTemplate rocketMQTemplate;


    //向mq发送新增用户消息
    @Override
    public void createAccountMessage(AccountChangeEvent accountChangeEvent) {

        //将accountChangeEvent转成json
        JSONObject jsonObject =new JSONObject();
        jsonObject.put("createAccount",accountChangeEvent);
        String jsonString = jsonObject.toJSONString();
        //生成message类型
        Message<String> message = MessageBuilder.withPayload(jsonString).build();
        //发送一条事务消息
        /**
         * String txProducerGroup 生产组
         * String destination topic，
         * Message<?> message, 消息内容
         * Object arg 参数
         */
        rocketMQTemplate.sendMessageInTransaction("producer_group_txmsg","topic_txmsg",message,null);

    }

    //创建用户
    @Override
    @Transactional
    public void createAccount(AccountChangeEvent accountChangeEvent) {
        //幂等判断
        if(accountInfoDao.isExistTx(accountChangeEvent.getTxNo())>0){
            return ;
        }
        //创建用户
        accountInfoDao.createAccount(accountChangeEvent.getAccountNo());
        //添加事务日志
        accountInfoDao.addTx(accountChangeEvent.getTxNo());
    }
}

3.执行本地事务

当我们发送了半消息之后，会回调ProducerTxmsgListener#executeLocalTransaction，因为我做了监听@RocketMQTransactionListener

这个时候我们可以拿到发送出去的message，执行上面本地事务accountInfoService#createAccount，保存用户数据，并且保存事务日志（插入txNo（事务id））。

4.事务回查

ProducerTxmsgListener#checkLocalTransaction 查询本地事务是否完成

由于我们Message里面有一个txNo事务id，当我们执行ProducerTxmsgListener#executeLocalTransaction的时候我们会告诉rocketmq，这次本地事务是成功还是失败，如果成功了会插入事务日志表，失败就没有这条数据。

由于网络不稳定，我们告诉rocketmq的消息他没收到，rocketmq就会带着消息体定时调用我们的

ProducerTxmsgListener#checkLocalTransaction，我们从里面拿出txNo,查询当前事务是否执行完成，如果执行完成，就返回commit，消息可以对消费者可见。

@Component
@Slf4j
@RocketMQTransactionListener(txProducerGroup = "producer_group_txmsg")
public class ProducerTxmsgListener implements RocketMQLocalTransactionListener {

    @Autowired
    AccountInfoService accountInfoService;

    @Autowired
    AccountInfoDao accountInfoDao;

    //事务消息发送后的回调方法，当消息发送给mq成功，此方法被回调
    @Override
    @Transactional
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {

        try {
            //解析message，转成AccountChangeEvent
            String messageString = new String((byte[]) message.getPayload());
            JSONObject jsonObject = JSONObject.parseObject(messageString);
            String accountChangeString = jsonObject.getString("accountChange");
            //将accountChange（json）转成AccountChangeEvent
            AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
            //执行本地事务，创建用户
            accountInfoService.createAccount(accountChangeEvent);
            //当返回RocketMQLocalTransactionState.COMMIT，自动向mq发送commit消息，mq将消息的状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            return RocketMQLocalTransactionState.ROLLBACK;
        }


    }

    //事务状态回查，查询是否执行成功
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析message，转成AccountChangeEvent
        String messageString = new String((byte[]) message.getPayload());
        JSONObject jsonObject = JSONObject.parseObject(messageString);
        String accountChangeString = jsonObject.getString("accountChange");
        //将accountChange（json）转成AccountChangeEvent
        AccountChangeEvent accountChangeEvent = JSONObject.parseObject(accountChangeString, AccountChangeEvent.class);
        //事务id
        String txNo = accountChangeEvent.getTxNo();
        int existTx = accountInfoDao.isExistTx(txNo);
        if(existTx>0){
            return RocketMQLocalTransactionState.COMMIT;
        }else{
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }
}

PointClient 积分服务

这里就不再写代码了，消费端和本地消息表的代码一模一样。注意好幂等性就好了，新增完积分然后给mq返回ack。

注意如果消费一直失败，需要预警 人工处理。

3.3.3 可靠消息最终一致总结

好处：

RocketMQ主要解决了两个功能：

1.本地事务与消息发送的原子性问题。

2.事务参与方接收消息的可靠性。 可靠消息最终一致性事务适合执行周期长且实时性要求不高的场景。引入消息机制后，同步的事务操作变为基于消息执行的异步操作, 避免了分布式事务中的同步阻塞操作的影响，并实现了两个服务的解耦。

3。性能比本地消息表好，能承受并发更高

坏处：

如果用rocketmq事务消息，强依赖了rocketmq；我们自己也可以按照rocketmq的这种自己实现一个可靠消息服务，但是增加了研发成本和系统复杂度。

3.4?最大努力通知

我们平常在接入微信支付的时候，只要用户支付成功了，微信会已一定的频率回调我们，假如我们回复微信失败了，他会过一段时间再通知，最多通知15次。

15次之后，我们就只能主动去微信那边查询这笔订单是否成功。微信的这个服务就是最大努力通知

3.4.1 最大努力通知实现方案

由于这里的代码比较简单，就是简化版的可靠消息，这里就不写了，只列出实现方案。

?3.4.2 最大努力通知和可靠性消息的异同

1.解决方案思想不同

--可靠消息一致性，发起通知方需要保证将消息发出去，并且将消息发到接收通知方，消息的可靠性关键由发起通知方来保证。

3.5 业务性分布式事务（日志加延时消息）

    @Transactional
    public Result createOrder(Product product){
        //生成订单编号
        String orderNo= RandomUtil.ranDomStr(5);
        DeductProduct decute=new DeductProduct();
        decute.setCount(product.getCount);
        decute.setProductId(product.getId);
        decute.setOrderNo(orderNo);
        //feign调用库存服务减库存
        boolean success=storeFeignClient.decute(decute);
        if(!success){
            return "error";
        }
        //feign给用户增加积分
        Point point=new Point();
        point.setPoint(10);
        point.setOrderNo(orderNo);
        //feign调用库存服务减库存
        boolean success1=pointFeignClient.addPOint(point);
        if(!success1){
            return "error";
        }
        //插入本地事务订单
        Order order=new Order();
        BeanUtils.copyProperties(product,order);
        order.setOrderNo(orderNo);
        OrderMapper.insert(order);
    }

3.5.1 业务性分布式事务总结

3.6 同步双写（多写）+异步对账

3.6.2 同步双写事务代码落地

3.6.1?同步双写事务解决方案

3.6.2?同步双写（多写）+异步对账总结

3.7 阿里巴巴Seata框架

?3.7.1 seataAT常见问题与XA对比

1.全局事务还没完成，数据被其他事务修改

假设一个请求基于 seata 对一条数据加了全局锁，并进行了本地分支事务提交。此时全局事务没有提交，另一个请求过来（这个请求就是简单的更新这条数据，不是分布式事务）自然就不需要获取全局锁，也就是可以直接更新。那全局事务再进行回滚时是不是就有问题了？

没错，描述的这个场景是有问题的，这也正是我们开发时需要考虑的地方，当有可能操作同一条数据引起并发问题的时候，给他加上全局事务，去获取全局锁。

2.既然seata也有全局锁，那么和我XA有什么区别，大家一阶段都是锁资源。

架构层次方面，传统2PC方案的 RM 实际上是在数据库层，RM 本质上就是数据库自身，通过 XA 协议实现，而 Seata的 RM 是以jar包的形式作为中间件层部署在应用程序这一侧的。 两阶段提交方面，传统2PC无论第二阶段的决议是commit还是rollback，事务性资源的锁都要保持到Phase2完成 才释放。而Seata的做法是在Phase1 就将本地事务提交，这样就可以省去Phase2持锁的时间，整体提高效率。

简单来说，就是seataAT模式一阶段已经提交本地事务了，不会再持有数据库资源，但是别人想修改，必须先获取全局锁（所需要修改数据Id的集合）。而XA模式只要你事务一天不提交，我数据库就他么锁着这些数据，谁也不可以动，本人亲测，在XA执行的时候抛出异常，你一天没提交事务，那一行就这么被锁着，不能修改，如果修改的数据没有条件，就是表锁，是表锁就等着爆炸吧。所以seata at模式和我们2pc的XA很像，只不过seata是在一阶段先提交了数据，后续根据undo_log补偿。

3.什么是全局锁

假如我的用户13下单操作，需要锁商品id为1和id=5的库存，需要给用户新增积分，需要修改用户的余额，那么我们会把商品库存表id=1和id=5，用户表的id=13，全部收集起来，用tableName+id作为唯一索引，插入数据库，商品库存表锁了2个id就插入两条数据。（新增的数据不会有全局锁）

如果其他事物想去修改对应的数据，就必须先去获取全局锁，看看我这次修改的id在不在全局锁，如果在就在一段时间内循环获取全局锁。? ? ? ??

4.假如全局事务回滚，他会根据你的Undo_log日志去补偿之前的数据，假如是update操作他发现，undo_log存的数据和现在的数据不一致（所以需要全局锁），则会异常，需要人工补偿。

5.用了seata at模式，我们就不能随意update数据库的字段了，不然会死得很惨。

3.7.2 代码落地（下一篇博文会有详细介绍）

实际上seata AT模式是无侵入的，只需要加一个注解

1、全局事务开始使用 @GlobalTransactional标识 。

2、每个本地事务方案仍然使用@Transactional标识。

3、每个数据都需要创建undo_log表，此表是seata保证本地事务一致性的关键。

4、不管你本地事务还是分布式事务，只要你业务需要分布式事务加了@GlobalTransactional，那么本地事务也得加这个@GlobalTransactional，去获取全局锁，假如我的下单，需要锁库存，新增积分，我给这个流程加了@GlobalTransactional。那么假如用户退款，我们需要修改订单状态，这个操作不需要分布式事务，只需要本地事务，我们也得加@GlobalTransactional注解，他要去获取全局锁，没有全局锁才可以操作。

5、@GlobalTransactional这个注解有点重，他会http去注册全局事务和分支事务，再去获取全局锁，如果我们只有本地事务的存在的时候可以用@GlobalLock去获取全局锁，他不会去注册全局和分支事务，只会去获取全局锁。

3.7.3 seata AT模式总结

优点：业务无侵入，我们只需要加一个注解就可以了，原理都是代理数据源，增强commit等方法。

缺点：全局锁如果没有加好，其他事务在全局事务没执行完就去修改数据，那就等着原地爆炸吧。

比较适合一些数据只有少数人可以修改的场景，例如我们的后台服务，对应的人只能操作自己的数据。而且每一步关系到这些表的数据修改操作，都得去获取全局锁，造成性能问题。

3.8 saga事务

其实这个模式很少用，我们来看一下他的简单介绍。
TCC事务，其实就是将一个接口拆分为3个接口，try -> confirm -> cancel，try成功了就confirm，try失败了就cancel。那如果confirm或者cancel失败了呢？confirm/cancel失败了就是一直重试。

saga是将每个接口，拆分为2个接口，一个是业务接口，另外一个是补偿接口，相当于就是说将tcc里面的try和confirm合并为一个接口，就是先执行业务接口，直接就尝试完成整个业务逻辑的操作然后如果在服务调用链条里面，某个服务的业务接口执行失败了，那么直接对已经执行成功的所有服务都调用其补偿接口，将之前执行成功的业务逻辑给回滚。
saga这个东西，其实核心和本质，就是把每个操作分为实际的业务逻辑以及补偿业务逻辑，正常情况下，就依次执行各个服务的业务逻辑就好了，如果某个服务调用失败的话，直接对之前执行成功的那些服务都依次执行补偿逻辑。
saga事务的思想，就是将一个长的分布式事务，拆分为一连串的每个服务的本地事务，然后每个服务对每个接口提供两个接口，一个是业务接口，一个是回滚的补偿接口，正常情况下就是依次的进行调用。
异常情况下，就对之前已经执行成功的服务执行补偿接口，回滚业务逻辑。

4.分布式事务选型