聊聊Seata分布式事务解决方案的AT模式

Posted 又见阿郎

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了聊聊Seata分布式事务解决方案的AT模式相关的知识,希望对你有一定的参考价值。

什么是Seata分布式事务解决方案

Seata是一款开源的分布式事务解决方案,致力于提供高性能和简单易用的分布式事务服务。若把经典的TCC 模式类比成手动挡驾驶模式,那么 Seata 的 AT 模式就好比自动挡驾驶模式,分布式事务这个强大且复杂的服务能力由Seata框架托管,对业务实现无侵入式,用户仍然只专注于业务 SQL。

AT模式

AT 模式是基于 DataSource 代理实现的,通过代理 DataSource ,拦截 SQL 执行,增强其执行逻辑,由代理侧加入额外的能力以提供分布式事务服务(好似钢铁侠的机甲提供了超强能力)。

暂时先放个自己整理的脑图概要,后面会铺开细节。

Seata架构篇 - AT模式

前言

Seata 是一款开源的分布式事务解决方案,致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式,为用户打造一站式的分布式解决方案。

X/Open DTP Model

X/Open DTP Model 是 X/Open 组织定义的一套分布式事务的标准,也就是定义了分布式事务处理的规范和 API 接口,但是具体实现由各厂商自行实现。

TM、RM 作为 Seata 的客户端与业务系统集成在一起,TC 作为 Seata 的服务端独立部署。

TC(Transaction Coordinator)- 事务协调者

维护全局和分支事务的状态,驱动全局事务提交或回滚。

TM(Transaction Manager)- 事务管理器

管理全局事务,包括开始全局事务、提交或回滚全局事务。

RM(Resource Manager)- 资源管理器

管理分支事务,包括向 TC 注册分支事务和报告分支事务的状态,并驱动分支事务提交或回滚。

在 Seata 中,分布式事务的执行流程:

  • TM 开启分布式事务(TM 向 TC 注册全局事务记录)
  • 按业务场景,编排数据库、服务等事务资源(RM 向 TC 汇报资源准备状态)
  • TM 结束分布式事务,事务阶段一结束(TM 通知 TC 提交/回滚分布式事务)
  • TC 汇总事务信息,决定分布式事务提交还是回滚
  • TC 通知所有 RM 提交/回滚,事务二阶段结束

AT 模式

概述

Seata AT 模式是一种非侵入式的分布式事务解决方案,Seata 在内部做了对数据库操作的代理层,我们使用 Seata AT 模式时,实际上用的是 Seata 自带的数据源代理 DataSourceProxy,Seata 在这层代理中加入了很多逻辑,比如插入回滚 undo_log 日志,检查全局锁等。

为什么要检查全局锁呢,这是由于 Seata AT 模式的事务隔离是建立在分支事务的本地隔离级别基础之上的,在数据库本地隔离级别读已提交或以上的前提下,Seata 设计了由事务协调器维护的全局写排他锁,来保证事务间的写隔离,同时,将全局事务默认定义在读未提交的隔离级别上。

Seata 的事务是一个全局事务,它包含了若干个分支本地事务,在全局事务执行过程中(全局事务还没执行完),某个本地事务提交了,如果 Seata 没有采取任何措施,则会导致已提交的本地事务被读取,造成脏读,如果数据在全局事务提交前已提交的本地事务被修改,则会造成脏写。

一阶段:业务数据和回滚日志在同一个本地事务提交,释放本地锁和连接资源。

二阶段:提交异步化,非常快速地完成。回滚通过一阶段的回滚日志反向补偿。

Seata AT模式隔离级别解读

适用场景

  • 基于支持本地 ACID 事务的关系型数据库。
  • Java 应用,通过 JDBC 访问数据库。

原理

一阶段

1、解析 SQL。得到 SQL 的类型(UPDATE)、表(product)、条件(where name = ‘TXC’)等相关的信息。

2、查询前镜像:根据解析得到的条件信息,生成查询语句,定位数据。

select id, name, since from product where name = 'TXC';

得到前镜像:

idnamesince
1TXC2014

3、执行业务 SQL。更新这条记录的 name 为 ‘GTS’。

4、查询后镜像:根据前镜像的结果,通过 主键 定位数据。

得到后镜像:

idnamesince
1GTS2014

5、插入回滚日志:把前后镜像数据以及业务 SQL 相关的信息组成一条回滚日志记录,插入到 UNDO_LOG 表中。


	"branchId": 641789253,
	"undoItems": [
		"afterImage": 
			"rows": [
				"fields": [
					"name": "id",
					"type": 4,
					"value": 1
				, 
					"name": "name",
					"type": 12,
					"value": "GTS"
				, 
					"name": "since",
					"type": 12,
					"value": "2014"
				]
			],
			"tableName": "product"
		,
		"beforeImage": 
			"rows": [
				"fields": [
					"name": "id",
					"type": 4,
					"value": 1
				, 
					"name": "name",
					"type": 12,
					"value": "TXC"
				, 
					"name": "since",
					"type": 12,
					"value": "2014"
				]
			],
			"tableName": "product"
		,
		"sqlType": "UPDATE"
	],
	"xid": "xid:xxx"

6、提交前,向 TC 注册分支:申请 product 表中,主键值等于 1 的记录的 全局锁

7、本地事务提交:业务数据的更新和前面步骤中生成的 UNDO LOG 一并提交。

8、将本地事务提交的结果上报给 TC。

二阶段

回滚

1、收到 TC 的分支回滚请求,开启一个本地事务,执行如下操作。

2、通过 XID 和 Branch ID 查找到相应的 UNDO LOG 记录。

3、数据校验:拿 UNDO LOG 中的后镜与当前数据进行比较,如果有不同,说明数据被当前全局事务之外的动作做了修改。这种情况,需要根据配置策略来做处理,详细的说明在另外的文档中介绍。

4、根据 UNDO LOG 中的前镜像和业务 SQL 的相关信息生成并执行回滚的语句:

update product set name = 'TXC' where id = 1;

5、提交本地事务。并把本地事务的执行结果(即分支事务回滚的结果)上报给 TC。

提交

1、收到 TC 的分支提交请求,把请求放入一个异步任务的队列中,马上返回提交成功的结果给 TC。

2、异步任务阶段的分支提交请求将异步和批量地删除相应 UNDO LOG 记录。

回滚日志表如下:

CREATE TABLE `undo_log` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) NOT NULL,
  `context` varchar(128) NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime NOT NULL,
  `log_modified` datetime NOT NULL,
  PRIMARY KEY (`id`),
  UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

写隔离

一阶段本地事务提交之前,需要确保先拿到全局锁,这样才能提交本地事务。

尝试获取全局锁如果超时,则放弃尝试,然后回滚本地事务并释放本地锁。

两个全局事务 tx1 和 tx2,分别对 a 表的 m 字段进行更新操作,m 的初始值 1000。tx1 先开始,开启本地事务,拿到本地锁,更新操作 m = 1000 - 100 = 900。本地事务提交前,先拿到该记录的 全局锁 ,本地提交释放本地锁。 tx2 后开始,开启本地事务,拿到本地锁,更新操作 m = 900 - 100 = 800。本地事务提交前,尝试拿该记录的 全局锁 ,tx1 全局提交前,该记录的全局锁被 tx1 持有,tx2 需要重试等待 全局锁

tx1 二阶段全局提交,释放 全局锁 。tx2 拿到 全局锁 提交本地事务。

如果 tx1 的二阶段全局回滚,则 tx1 需要重新获取该数据的本地锁,进行反向补偿的更新操作,实现分支的回滚。此时,如果 tx2 仍在等待该数据的 全局锁,同时持有本地锁,则 tx1 的分支回滚会失败。分支的回滚会一直重试,直到 tx2 的 全局锁 等锁超时,放弃 全局锁 并回滚本地事务释放本地锁,tx1 的分支回滚最终成功。

因为整个过程 全局锁 在 tx1 结束前一直是被 tx1 持有的,所以不会发生 脏写 的问题。

读隔离

在数据库本地事务隔离级别 读已提交(Read Committed) 或以上的基础上,Seata(AT 模式)的默认全局隔离级别是 读未提交(Read Uncommitted)

SELECT FOR UPDATE 语句的执行会申请 全局锁 ,如果 全局锁 被其他事务持有,则释放本地锁(回滚 SELECT FOR UPDATE 语句的本地执行)并重试。这个过程中,查询是被 block 住的,直到 全局锁 拿到,即读取的相关数据是 已提交 的,才返回。

出于总体性能上的考虑,Seata 目前的方案并没有对所有 SELECT 语句都进行代理,仅针对 FOR UPDATE 的 SELECT 语句。

使用

假设 Bussiness 服务调用 Stock 服务和 Order 服务。

1、配置 Bussiness 服务。

  • 配置 GlobalTransactionScanner 的applicationId 和 txServiceGroup(事务分组)。

  • 使用 @GlobalTransactional 注解

  • 可以使用 RootContext.getXID() 方法获取到 XID

  • 按需修改 file.conf、registry.conf 文件的内容

@Configuration
public class SeataAutoConfig 

    @Bean
    public GlobalTransactionScanner globalTransactionScanner() 
      	// 指定 applicationId、txServiceGroup
        return new GlobalTransactionScanner("dubbo-gts-seata-example", "my_test_tx_group");
    

@Service
public class BusinessServiceImpl implements BusinessService 

    @Reference(version = "1.0.0")
    private StockDubboService stockDubboService;

    @Reference(version = "1.0.0")
    private OrderDubboService orderDubboService;

    private boolean flag;

    @Override
    @GlobalTransactional(timeoutMills = 300000, name = "dubbo-gts-seata-example")
    public ObjectResponse handleBusiness(BusinessDTO businessDTO) 
        System.out.println("开始全局事务,XID = " + RootContext.getXID());
        ObjectResponse<Object> objectResponse = new ObjectResponse<>();
        //1、扣减库存
        CommodityDTO commodityDTO = new CommodityDTO();
        commodityDTO.setCommodityCode(businessDTO.getCommodityCode());
        commodityDTO.setCount(businessDTO.getCount());
        ObjectResponse stockResponse = stockDubboService.decreaseStock(commodityDTO);
        //2、创建订单
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setUserId(businessDTO.getUserId());
        orderDTO.setCommodityCode(businessDTO.getCommodityCode());
        orderDTO.setOrderCount(businessDTO.getCount());
        orderDTO.setOrderAmount(businessDTO.getAmount());
        ObjectResponse<OrderDTO> response = orderDubboService.createOrder(orderDTO);

        //打开注释测试事务发生异常后,全局回滚功能
        //  if (!flag) 
        //      throw new RuntimeException("测试抛异常后,分布式事务回滚!");
        //  

        if (stockResponse.getStatus() != 200 || response.getStatus() != 200) 
            throw new DefaultException(RspStatusEnum.FAIL);
        

        objectResponse.setStatus(RspStatusEnum.SUCCESS.getCode());
        objectResponse.setMessage(RspStatusEnum.SUCCESS.getMessage());
        objectResponse.setData(response.getData());
        return objectResponse;
    

2、 配置 Stock 服务。

  • 配置 DataSource、DataSourceProxy、SqlSessionFactory、GlobalTransactionScanner
  • 可以使用 RootContext.getXID() 方法获取到 XID
  • 按需修改 file.conf、registry.conf 文件的内容
  • 对应数据库配置 UNDO_LOG 表
@Configuration
public class SeataAutoConfig 

    @Autowired
    private DataSourceProperties dataSourceProperties;

    @Bean
    @Primary
    public DruidDataSource druidDataSource() 
        DruidDataSource druidDataSource = new DruidDataSource();
        druidDataSource.setUrl(dataSourceProperties.getUrl());
        druidDataSource.setUsername(dataSourceProperties.getUsername());
        druidDataSource.setPassword(dataSourceProperties.getPassword());
        druidDataSource.setDriverClassName(dataSourceProperties.getDriverClassName());
        druidDataSource.setInitialSize(0);
        druidDataSource.setMaxActive(180);
        druidDataSource.setMaxWait(60000);
        druidDataSource.setMinIdle(0);
        druidDataSource.setValidationQuery("Select 1 from DUAL");
        druidDataSource.setTestOnBorrow(false);
        druidDataSource.setTestOnReturn(false);
        druidDataSource.setTestWhileIdle(true);
        druidDataSource.setTimeBetweenEvictionRunsMillis(60000);
        druidDataSource.setMinEvictableIdleTimeMillis(25200000);
        druidDataSource.setRemoveAbandoned(true);
        druidDataSource.setRemoveAbandonedTimeout(1800);
        druidDataSource.setLogAbandoned(true);
        return druidDataSource;
    

    @Bean
    public DataSourceProxy dataSourceProxy(DruidDataSource druidDataSource) 
        return new DataSourceProxy(druidDataSource);
    

    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSourceProxy dataSourceProxy) throws Exception 
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(dataSourceProxy);
        factoryBean.setMapperLocations(
            new PathMatchingResourcePatternResolver().getResources("classpath*:/mapper/*.xml"));
        factoryBean.setTransactionFactory(new JdbcTransactionFactory());
        return factoryBean.getObject();
    

    @Bean
    public GlobalTransactionScanner globalTransactionScanner() 
        return new GlobalTransactionScanner("stock-gts-seata-example", "my_test_tx_group");
    

3、配置 Order 服务

  • 配置 DataSource、DataSourceProxy、SqlSessionFactory、GlobalTransactionScanner
  • 可以使用 RootContext.getXID() 方法获取到 XID
  • 按需修改 file.conf、registry.conf 文件的内容
  • 对应数据库配置 UNDO_LOG 表
@Configuration
public class SeataAutoConfig 

    @Autowired
    private DataSourceProperties dataSourceProperties;

    @Bean
    @Primary
    public DruidDataSource druidDataSource() 
        DruidDataSource druidDataSource = new DruidDataSource();
        druidDataSource.setUrl(dataSourceProperties.getUrl());
        druidDataSource.setUsername(dataSourceProperties.getUsername());
        druidDataSource.setPassword(dataSourceProperties.getPassword());
        druidDataSource.setDriverClassName(dataSourceProperties.getDriverClassName());
        druidDataSource.setInitialSize(0);
        druidDataSource.setMaxActive(180);
        druidDataSource.setMaxWait(60000);
        druidDataSource.setMinIdle(0);
        druidDataSource.setValidationQuery("Select 1 from DUAL");
        druidDataSource.setTestOnBorrow(false);
        druidDataSource.setTestOnReturn(false);
        druidDataSource.setTestWhileIdle(true);
        druidDataSource.setTimeBetweenEvictionRunsMillis(60000);
        druidDataSource.setMinEvictableIdleTimeMillis(25200000);
        druidDataSource.setRemoveAbandoned(true);
        druidDataSource.setRemoveAbandonedTimeout(1800);
        druidDataSource.setLogAbandoned(true);
        return druidDataSource;
    

    @Bean
    public DataSourceProxy dataSourceProxy(DruidDataSource druidDataSource) 
        return new DataSourceProxy(druidDataSource);
    

    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSourceProxy dataSourceProxy) throws Exception 
        SqlSessionFactoryBean factoryBean = new SqlSessionFactoryBean();
        factoryBean.setDataSource(dataSourceProxy);
        factoryBean.setMapperLocations(
            new PathMatchingResourcePatternResolver().getResources("classpath*:/mapper/*.xml"));
        factoryBean.setTransactionFactory(new JdbcTransactionFactory());
        return factoryBean.getObject();
    

    @Bean
    public GlobalTransactionScanner globalTransactionScanner() 
        return new GlobalTransactionScanner("order-gts-seata-example", "my_test_tx_group");
    

以上是关于聊聊Seata分布式事务解决方案的AT模式的主要内容,如果未能解决你的问题,请参考以下文章

深入解析分布式事务中间件seata-AT模式

seata 分布式事务框架seata1.3 AT及XA模式实例演示

Seata架构篇 - AT模式

分布式事务 - 分布式事务框架Seata

Seata-AT 如何保证分布式事务一致性

分布式事务 - AT模式Dubbo集成Seata