Seata 分布式事务解决方案：Dubbo 与 Spring Cloud 集成实践

一、背景介绍

在之前的文章中，我们简单介绍了一下 Spring Cloud Alibaba 的技术体系中的 Nacos、Dubbo 和 Sentinel 组件应用，通过这几款组件基本可以构建一个简易版的微服务框架系统。

我们知道，在微服务系统中一些模块通常会以一个独立的服务来开发和部署，比如用户服务、订单服务、库存服务、账单服务等等。随着服务拆分的越来越细，微服务的数量也会随之增长，系统的复杂度也会变得很高。例如，当用户选择某个商品下一笔单的时候，通常会先调用库存服务的库存扣减逻辑，如果库存充足，接着再调用订单服务的创建订单逻辑。看似一个简单的操作，其实至少会经过两个微服务的数据写入动作。从业务角度来看，这一系列的服务操作，要么一起成功，要么一起失败，才能保障系统数据的完整性。

众所周知，在微服务系统中，不同的服务通常对应的数据库也不一样，因此当涉及到多个服务的数据操作时，必然会存在跨库提交数据的现象。

当请求一切都正常的时候，还好说；但是当某个服务节点出现异常时，就不好说了。比如调用库存服务的库存扣减逻辑成功了，但再调用订单服务的创建订单逻辑失败了，实际上这个订单没有创建成功，但是库存已经成功扣减了，从业务角度来看，库存数据显然并不正确。当创建订单时如果出现失败，订单服务需要反向调用库存增加逻辑将扣减的库存进行回滚，以便保证系统数据的正确性。

以上还只是介绍了两个服务的数据操作，而实际上在业务开发过程中，有的请求操作可能涉及到好几个微服务的数据写入动作，如果最后一个服务节点出现异常，意味着前面执行过的服务都得进行类似的数据回滚操作，通过人工写异常代码进行数据回滚显然很鸡肋。

在微服务环境下，有没有一种工具能帮助我们解决在跨库提交时数据不一致的问题呢？

答案肯定是有的，Spring Cloud Alibaba 技术生态中的 Seata 组件就可以帮忙我们解决这类问题。

二、Seata 简介

什么是 Seata？

Seata 是一款开源的分布式事务解决方案，由阿里巴巴集团开发并开源，旨在解决微服务架构下的分布式事务问题。它提供了高性能和易用性，同时支持多种事务模式，能帮助开发者在分布式系统中实现数据一致性。

下面我们一起来简单的了解一下它的架构设计。

2.1、Seata 架构图

在 Seata 的架构中，一共有三个角色：

各个角色承担着不同的用途：

TC (Transaction Coordinator) ：也被称为事务协调者，负责维护全局和分支事务的状态，驱动全局事务提交或回滚。
TM (Transaction Manager) ：也被称为事务管理器，负责定义全局事务的范围，开始全局事务、提交或回滚全局事务。
RM ( Resource Manager ) ：也被称资源管理器，负责管理分支事务处理的资源( Resource )，与 TC 交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚。

其中，TC 为单独部署的Server 服务端，TM 和 RM 为嵌入到应用中的 Client 客户端。

在 Seata 中，一个分布式事务的生命周期可以用如下图来概括：

具体的执行过程，可以用如下几个步骤来概括：

第一步：当某个服务开启分布式事务操作时，TM 会向 TC 发起请求开启一个全局事务，TC 会生成一个 XID 作为该全局事务的编号并返回给 TM。此时 XID 会在微服务的调用链路中传播，保证将多个微服务的子事务关联在一起。
第二步：接着 RM 会向 TC 发起请求将本地事务注册为全局事务的一个分支事务，同时通过全局事务的 XID 将其关联。
第三步：当服务执行完成以后，TM 会向 TC 发起请求告诉 XID 对应的全局事务是进行提交还是回滚操作。
第四步：最后，TC 会向 RM 们发起请求将 XID 对应的自己的本地事务进行提交还是回滚操作。
第五步：如果其中有任何一个分支事务操作出现了异常，TC 会将其记录下来，以便于人工介入。

2.2、事务模式

为了打造一站式的分布事务解决方案，Seata 为用户提供了 AT、TCC、SAGA 和 XA 四种事务模式。

目前使用的流行度情况是：AT > TCC > Saga。因此在学习 Seata 的时候，我们可以重点关注一下 AT 模式，搞懂背后的实现原理即可。

下面我们简单的介绍一下 AT 模式实现原理，其它的模式实现大家可以自行参阅官方Seata 文档。

2.2.1、AT 模式实现原理

AT 模式是一种无侵入的分布式事务解决方案。在 AT 模式下，每个数据库会被当做是一个 Resource，业务通过 JDBC 标准接口访问数据库资源时，Seata 框架会对所有请求进行拦截，并将业务数据和回滚日志记录到本地数据库，以便对当前事务的执行情况做出相应的处理。

与其它的事务模式相比，AT 模式实现原理相对来说要简单很多，而且不易出错，并且使用上也非常的简洁，只需要在对应的方法上添加@GlobalTransactional全局事务注解就可以开启分布式事务操作，示例如下：

// 开启一个全局事务，方法内的跨库数据操作要么全部成功，要么全部失败
@GlobalTransactional
public void purchase() {
    // 调用服务A
    serviceA.doSomething();
    // 调用服务B
    serviceB.doSomething();
}

在 AT 模式中，整个事务执行过程，可以用两个阶段来概括。

一阶段：业务 SQL 执行操作；
二阶段：Seata 框架根据一阶段执行情况自动进行事务的提交和回滚操作；

详细的执行流程如下！

2.2.1.1、一阶段流程

在一阶段，Seata 会拦截“业务 SQL”，解析 SQL 语义，找到“业务 SQL”要更新的业务数据。在业务数据被更新前，将其保存成“before image”，然后执行“业务 SQL”，在业务数据更新之后，再将其保存成“after image”，并将业务数据和回滚日志记录到本地日志表中。

以上操作全部在一个数据库事务内完成，因此一阶段操作的原子性可以得到保证。

具体执行流程可以用如下图来概括。

2.2.1.2、二阶段 - 提交流程

在二阶段，如果是提交的话，因为“业务 SQL”在一阶段已经提交至数据库，所以 Seata 框架只需将一阶段保存的快照数据删掉，完成数据清理即可。

2.2.1.3、二阶段 - 回滚流程

在二阶段，如果是回滚的话，Seata 就需要回滚一阶段已经执行的“业务 SQL”，还原业务数据，回滚方式便是用“before image”还原业务数据。但在还原前要首先要校验脏写，对比“数据库当前业务数据”和 “after image”，如果两份数据完全一致就说明没有脏写，可以还原业务数据，如果不一致就说明有脏写，出现脏写就需要转人工处理。

具体执行流程可以用如下图来概括。

AT 模式的一阶段、二阶段均由 Seata 框架自动生成，用户只需编写“业务 SQL”，便能轻松接入分布式事务，因此大家普遍认为 AT 模式是一种对业务无任何侵入的分布式事务解决方案，就像开启本地事务操作一样简单。

介绍了这么多，如何在项目中使用 Seata 呢？下面我们一起来看看具体的实践方式。

三、Seata 服务端部署

在使用 Seata 之前，我们需要先部署 Seata TC Server 服务，通过它来维护全局事务和分支事务的状态等信息。

具体的部署方式如下。

3.1、单机部署

访问https://github.com/seata/seata/releases，下载想要的 Seata 版本。

这里，我们选择v1.5.2版本来安装部署。

# 下载
$ wget https://github.com/seata/seata/releases/download/v1.5.2/seata-server-1.5.2.tar.gz
# 解压
$ tar -zxvf seata-server-1.5.2.tar.gz

解压之后，可以看到类似如下目录结构：

bin 目录存放的是 Seata 服务相关的启动脚本；
conf 目录存放的是 Seata 服务相关的配置项；
lib 目录存放的是 Seata 服务相关的依赖库；
log 目录存放的是 Seata 服务相关的启动日志；
script 目录存放的是 Seata 服务相关的配置案例脚本；

Seata TC Server 的启动方式非常简单，只需要在bin目录下执行对应的脚本，就可以启动服务。

mac/linux 系统，执行 sh seata-server.sh ，即可启动服务；
windows 系统，双击 seata-server.bat ，即可启动服务；

启动 Seata TC Server 后，默认的控制台访问端口是7091，在浏览器中访问http://127.0.0.1:7091，如果看到如下界面，说明启动成功了。

默认的用户名和密码为seata，登陆之后会看到如下界面。

如果无法访问，请确认服务是否启动成功，可以在log目录下查看相关的启动日志，排查具体的启动情况。

3.2、集群部署

默认配置下，Seata TC Server 使用的是 file 模式存储数据，全局事务会话信息在内存中读写，并持久化到本地文件中，数据读写性能较高，常用于学习或测试环境使用，不适合生产环境中部署。

对于生产环境，通常我们会采用 db 数据库来实现全局事务会话信息的共享，同时以集群的方式来部署，以便实现服务的高可用效果。

实现也很简单，以 Mysql 数据库为例，具体实现步骤如下。

3.2.1、数据持久化存储

首先，打开script目录，找到mysql.sql文件，它是一个数据库表初始化脚本，等会会用到它。

然后，在Mysql数据库中创建一个seata-server数据库，并在该库下执行mysql.sql脚本，最终结果如下图：

接着，在 Seata 安装包中打开conf/application.example.yml文件，找到store.db相关配置属性。

将其复制出来，然后拷贝到conf/application.yml文件中。

这里需要重点注意一下。

如果你的目标数据库是 Mysql 5.x，对应的驱动配置类为 com.mysql.jdbc.Driver ；
如果如果你的目标数据库是 Mysql 8.x，对应的驱动配置类应为 com.mysql.cj.jdbc.Driver ；

如果配置错误，可能会导致启动报错，连接不上目标数据源。

最后，重启 Seata TC Server 服务即可。

集群部署的方式也比较简单，将安装包部署在不同的机器上，共同连接一个目标数据源就可以了。

3.2.2、设置使用 Nacos 注册中心

Seata TC Server 对主流的注册中心也提供了集成，Seata 客户端可以通过注册中心获取 Seata TC Server 所在的服务实例。

引入注册中心之后，Seata 的交互流程可以用如下图来概括。

考虑到国内使用 Nacos 作为注册中心比较广泛，在这里我们简单的介绍一下它的配置方式。

与上文类似，在 Seata 安装包中打开conf/application.example.yml文件，找到store.registry相关配置属性。

将其复制出来，然后拷贝到conf/application.yml文件中。

最后，重启 Seata TC Server 服务即可。

访问 nacos 的服务控制台，如果看到 Seata 服务，说明服务注册成功了。

四、Seata 客户端应用

Seata 服务端部署完成后，下面我们一起来看看应用服务如何接入 Seata。

目前 Seata 对主流的服务远程调用框架都进行适配和支持，比如常用的 Dubbo、gRPC、Apache HttpClient、Spring Cloud OpenFeign、Spring RestTemplate 等，因此只需要在 SpringBoot 项目中引入seata-spring-boot-starter依赖包，Seata 客户端会自动集成到项目中。

因为 Seata 是通过 DataSource 数据源进行代理实现，所以天然对主流的 ORM 框架提供了非常好的支持，比如 JPA、MyBatis 等，当项目引入上文提到的 Seata 依赖包，Seata 会对服务进行自动装配处理，无需在引入额外的包。

下面我们以用户下单为例，先调用库存服务进行扣减库存，如果成功再调用订单服务进行创建订单，介绍一下如何使用 seata 来实现分布式事务提交和回滚操作。

服务之间的交互流程可以用如下图来简要概括。

项目中用到的核心组件及对应的版本如下：

Spring Boot：2.2.5.RELEASE
Spring Cloud Alibaba：2.2.1.RELEASE
Mybatis：3.5.0
Mysql：8.0
Seata：1.1.0
Http Client：4.5.11

具体的实践过程如下！

4.1、创建库存服务

4.1.1、初始化数据库

首先，创建一个seata-stock数据库，并初始化相关业务表，示例如下：

CREATE TABLEIFNOTEXISTS`tb_stock` (
`id`int(11) NOTNULL AUTO_INCREMENT,
`product_code`varchar(255) DEFAULTNULL,
`count`int(11) DEFAULT0,
  PRIMARY KEY (`id`),
UNIQUEKEY (`commodity_code`)
) ENGINE=InnoDBDEFAULTCHARSET=utf8;
INSERTINTO`tb_stock` (`id`, `product_code`, `count`)
VALUES (1, 'wahaha', 10);
CREATETABLEIFNOTEXISTS`undo_log`
(
    `branch_id`     BIGINT       NOTNULLCOMMENT'branch transaction id',
    `xid`           VARCHAR(128) NOTNULLCOMMENT'global transaction id',
    `context`       VARCHAR(128) NOTNULLCOMMENT'undo_log context,such as serialization',
    `rollback_info` LONGBLOB     NOTNULLCOMMENT'rollback info',
    `log_status`    INT(11)      NOTNULLCOMMENT'0:normal status,1:defense status',
    `log_created`   DATETIME(6)  NOTNULLCOMMENT'create datetime',
    `log_modified`  DATETIME(6)  NOTNULLCOMMENT'modify datetime',
    UNIQUEKEY`ux_undo_log` (`xid`, `branch_id`)
    KEY`ix_log_created` (`log_created`)
) ENGINE = InnoDB AUTO_INCREMENT = 1DEFAULTCHARSET = utf8mb4 COMMENT ='AT transaction mode undo table';

其中，每个库中必须包含undo_log表，它是 Seata AT 模式必须创建的表，主要用于分支事务的回滚。

4.1.2、创建服务应用

然后，建一个 Spring Boot 工程，命名为seata-client-order，并在pom.xml中引入相关的依赖内容，示例如下：

<properties>
    <spring-boot.version>2.2.5.RELEASE</spring-boot.version>
    <spring-cloud.version>Hoxton.SR3</spring-cloud.version>
    <spring-cloud-alibaba.version>2.2.1.RELEASE</spring-cloud-alibaba.version>
</properties>
<dependencies>
    <!-- SpringBoot web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <!--mysql 驱动-->
    <dependency>
        <groupId>mysql</groupId>
        <artifactId>mysql-connector-java</artifactId>
    </dependency>
    <!--mybatis-->
    <dependency>
        <groupId>org.mybatis.spring.boot</groupId>
        <artifactId>mybatis-spring-boot-starter</artifactId>
        <version>2.0.0</version>
    </dependency>
    <!--引入 seata 分布式事务组件 -->
    <dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    </dependency>
    <!-- 实现 Seata 对 HttpClient 的集成支持  -->
    <dependency>
        <groupId>io.seata</groupId>
        <artifactId>seata-http</artifactId>
        <version>1.1.0</version>
    </dependency>
    <!-- Apache HttpClient 依赖 -->
    <dependency>
        <groupId>org.apache.httpcomponents</groupId>
        <artifactId>httpclient</artifactId>
        <version>4.5.11</version>
    </dependency>
</dependencies>
<dependencyManagement>
    <dependencies>
        <!-- 引入 springBoot 版本号 -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>${spring-boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <!-- 引入 spring cloud 版本号 -->
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>${spring-cloud.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
        <!-- 引入 spring cloud alibaba 适配的版本号 -->
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>${spring-cloud-alibaba.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

其中引入的seata-spring-boot-starter依赖，会自动装载 Seata 客户端相关配置。

4.1.3、编写库存扣减接口

接着，创建库存扣减接口，以便用于发起调用，示例如下：

@RestController
@RequestMapping("/stock")
publicclass StockController {
    @Autowired
    private StockService stockService;
    @PostMapping("/deduct")
    public Boolean deduct(@RequestBody Stock stock) {
        try {
            stockService.deduct(stock.getProductCode(), stock.getCount());
            // 正常扣除库存，返回 true
            returntrue;
        } catch (Exception e) {
            // 失败扣除库存，返回 false
            returnfalse;
        }
    }
}

对应的service层代码如下：

@Component
publicclass StockService {
    @Autowired
    private StockMapper stockMapper;
    @Transactional
    public void deduct(String productCode, int count){
        Stock stock = new Stock();
        stock.setProductCode(productCode);
        stock.setCount(count);
        // 扣减库存
        stockMapper.update(stock);
    }
}

对应的dao层 SQL 代码如下：

<update id="update" parameterType="com.example.cloud.seata.client.entity.Stock">
    update tb_stock
    set `count` = `count` - #{count}
    where product_code = #{productCode}
</update>

4.1.4、创建服务启动类

最后，创建服务启动类。

@MapperScan("com.example.cloud.seata.client")
@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class,args);
    }
}

同时，创建全局配置文件并添加 seata 服务端相关的配置属性，示例如下。

spring.application.name=seata-client-stock
server.port=9002
# 添加数据源配置
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/seata-stock
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
# 配置mybatis全局配置文件扫描
mybatis.config-location=classpath:mybatis/mybatis-config.xml
# 配置mybatis的xml配置文件扫描目录
mybatis.mapper-locations=classpath:mybatis/mapper/*.xml
# 添加Seata 配置项
# Seata 应用编号，默认为spring.application.name
seata.application-id=seata-client-stock
# Seata 事务组编号，用于 TC 集群名
seata.tx-service-group=my_test_tx_group
# Seata 服务配置项，配置对应的虚拟组和分组的映射，其中127.0.0.1:8091为 seata 服务端的监听端口
seata.service.vgroup-mapping.my_test_tx_group=default
seata.service.grouplist.default=127.0.0.1:8091

以上工程完成之后，将服务启动起来。当看到如下信息，说明服务已经成功启动。

4.2、创建订单服务

4.2.1、初始化数据库

与上文类似，首先，创建一个seata-order数据库，并初始化相关业务表，示例如下：

CREATE TABLE`tb_order` (
`id`intNOTNULL AUTO_INCREMENT,
`user_id`varchar(255) DEFAULTNULL,
`product_code`varchar(255) DEFAULTNULL,
`count`intDEFAULT'0',
`money`intDEFAULT'0',
  PRIMARY KEY (`id`)
) ENGINE=InnoDBDEFAULTCHARSET=utf8;
CREATETABLEIFNOTEXISTS`undo_log`
(
    `branch_id`     BIGINT       NOTNULLCOMMENT'branch transaction id',
    `xid`           VARCHAR(128) NOTNULLCOMMENT'global transaction id',
    `context`       VARCHAR(128) NOTNULLCOMMENT'undo_log context,such as serialization',
    `rollback_info` LONGBLOB     NOTNULLCOMMENT'rollback info',
    `log_status`    INT(11)      NOTNULLCOMMENT'0:normal status,1:defense status',
    `log_created`   DATETIME(6)  NOTNULLCOMMENT'create datetime',
    `log_modified`  DATETIME(6)  NOTNULLCOMMENT'modify datetime',
    UNIQUEKEY`ux_undo_log` (`xid`, `branch_id`)
    KEY`ix_log_created` (`log_created`)
) ENGINE = InnoDB AUTO_INCREMENT = 1DEFAULTCHARSET = utf8mb4 COMMENT ='AT transaction mode undo table';

4.2.2、创建服务应用

然后，建一个 Spring Boot 工程，命名为seata-client-order，依赖包与上文完全一致，这里就不再重复粘贴了。

4.2.3、编写订单创建接口

接着，创建一个下单接口，示例如下：

@RestController
@RequestMapping("/order")
publicclass OrderController {
    privatestaticfinal Logger LOGGER = LoggerFactory.getLogger(OrderController.class);
    @Autowired
    private OrderService orderService;
    @GetMapping("/create")
    public String create(@RequestParam("userId") String userId,
                          @RequestParam("productCode") String productCode,
                          @RequestParam("orderCount") Integer orderCount) {
        try {
            orderService.create(userId, productCode, orderCount);
            return"订单创建成功！";
        } catch (Exception e) {
            LOGGER.error("错误信息", e);
            return"订单创建失败！";
        }
    }
}

对应的service层代码如下：

@Component
publicclass OrderService {
    @Autowired
    private OrderMapper orderMapper;
    /**
     * GlobalTransactional 表示当前服务需要开启分布式事务操作，当通过 http 发起远程调用的时候，seata 会将当前的全局事务会话 ID 传递到目标服务中。
     */
    @GlobalTransactional
    public void create(String userId, String productCode, int orderCount) throws Exception {
        // 扣减库存
        reduceStock(productCode, orderCount);
        Order order = new Order();
        order.setUserId(userId);
        order.setProductCode(productCode);
        order.setCount(orderCount);
        order.setMoney(orderCount * 100);
        // 创建订单
        orderMapper.insert(order);
    }
    /**
     * 通过 seata 包装的 HttpClient 工具发起服务远程调用
     */
    private void reduceStock(String productCode, Integer orderCount) throws IOException {
        // 参数拼接
        JSONObject params = new JSONObject()
                .fluentPut("productCode", productCode)
                .fluentPut("count", orderCount);
        // 执行调用
        HttpResponse response = DefaultHttpExecutor.getInstance().executePost("http://127.0.0.1:9002", "/stock/deduct", params, HttpResponse.class);
        // 解析结果
        Boolean success = Boolean.valueOf(EntityUtils.toString(response.getEntity()));
        if (!success) {
            thrownew RuntimeException("扣减库存失败");
        }
    }
}

对应的dao层 SQL 代码如下：

<insert id="insert" parameterType="com.example.cloud.seata.client.entity.Order">
    insert into tb_order(id, user_id, product_code, `count`, money)
    values(#{id}, #{userId}, #{productCode}, #{count}, #{money})
</insert>

可以发现，在OrderService类的create()方法中多了一个@GlobalTransactional，它表示当前服务需要开启分布式事务操作，当通过 http 发起远程调用的时候，seata 会将当前的全局事务会话 ID 传递到目标服务中。

其次，在发起服务远程调用时，需要用io.seata包中的Http工具来发起，因为它会将当前的全局事务会话 ID 信息作为头部参数，传递到目标服务中；如果使用 seata 不支持的组件，可能需要自行进行适配。

4.2.4、创建服务启动类

最后，与上文类似，创建服务启动类。

@MapperScan("com.example.cloud.seata.client")
@SpringBootApplication
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class,args);
    }
}

同时，创建全局配置文件并添加 seata 服务端相关的配置属性，示例如下。

spring.application.name=seata-client-order
server.port=9001
# 添加数据源配置
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/seata-order
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
# 配置mybatis全局配置文件扫描
mybatis.config-location=classpath:mybatis/mybatis-config.xml
# 配置mybatis的xml配置文件扫描目录
mybatis.mapper-locations=classpath:mybatis/mapper/*.xml
# 添加Seata 配置项
# Seata 应用编号，默认为spring.application.name
seata.application-id=seata-client-order
# Seata 事务组编号，用于 TC 集群名
seata.tx-service-group=my_test_tx_group
# Seata 服务配置项，配置对应的虚拟组和分组的映射，其中127.0.0.1:8091为 seata 服务端的监听端口
seata.service.vgroup-mapping.my_test_tx_group=default
seata.service.grouplist.default=127.0.0.1:8091