分布式事务(3)之seata实现2PC事务

男娘i 2022-10-16 06:25 250阅读 0赞

# 一、业务说明 #

本示例通过Seata中间件实现分布式事务，模拟三个账户的转账交易过程。两个账户在三个不同的银行(张三在bank1、李四在bank2)，bank1和bank2是两个个微服务。交易过程是，张三给李四转账指定金额。

上述交易步骤，要么一起成功，要么一起失败，必须是一个整体性的事务。

![20210527162930321.png][]

# 二、程序组成部分 #

本示例程序组成部分如下：

数据库：MySQL-5.7.25，包括bank1和bank2两个数据库。

JDK：64位 jdk1.8.0\_201

微服务框架：spring-boot-2.1.3、spring-cloud-Greenwich.RELEASE

seata客户端（RM、TM）：spring-cloud-alibaba-seata-2.1.0.RELEASE

seata服务端(TC)：seata-server-0.7.1

微服务及数据库的关系 ：

dtx/dtx-seata-demo/seata-demo-bank1 银行1，操作张三账户， 连接数据库bank1

dtx/dtx-seata-demo/seata-demo-bank2 银行2，操作李四账户，连接数据库bank2

服务注册中心：dtx/discover-server

本示例程序技术架构如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70][]\\

交互流程如下：

1、请求bank1进行转账，传入转账金额。

2、bank1减少转账金额，调用bank2，传入转账金额。

## 2.1创建数据库 ##

导入数据库脚本：资料\\sql\\bank1.sql、资料\\sql\\bank2.sql

包括如下数据库：bank1库，包含张三账户

CREATE DATABASE `bank1` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';

DROP TABLE IF EXISTS `account_info`;
    CREATE TABLE `account_info`  (
      `id` bigint(20) NOT NULL AUTO_INCREMENT,
      `account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '户
    主姓名',
      `account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '银行
    卡号',
      `account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT
    '帐户密码',
      `account_balance` double NULL DEFAULT NULL COMMENT '帐户余额',
      PRIMARY KEY (`id`) USING BTREE
    ) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT =
    Dynamic;
    INSERT INTO `account_info` VALUES (2, '张三的账户', '1', '', 10000);

bank2库，包含李四账户

CREATE DATABASE `bank2` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci';

CREATE TABLE `account_info`  (
      `id` bigint(20) NOT NULL AUTO_INCREMENT,
      `account_name` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '户
    主姓名',
      `account_no` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT '银行
    卡号',
      `account_password` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL COMMENT
    '帐户密码',
      `account_balance` double NULL DEFAULT NULL COMMENT '帐户余额',
      PRIMARY KEY (`id`) USING BTREE
    ) ENGINE = InnoDB AUTO_INCREMENT = 5 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT =
    Dynamic;
    INSERT INTO `account_info` VALUES (3, '李四的账户', '2', NULL, 0);

分别在bank1、bank2库中创建undo\_log表，此表为seata框架使用：

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,
      `ext` varchar(100) DEFAULT NULL,
      PRIMARY KEY (`id`),
      UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

## 2.2启动TC(事务协调器) ##

（1）下载seata服务器

下载地址：https://github.com/seata/seata/releases/download/v0.7.1/seata-server-0.7.1.zip,也可以直接解压：资料\\seata-server-0.7.1.zip

（2）解压并启动

\[seata服务端解压路径\]/bin/seata-server.bat -p 8888 -m file

注：其中8888为服务端口号；file为启动模式，这里指seata服务将采用文件的方式存储信息。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 1][]

## 2.3discover-server ##

discover-server是服务注册中心，测试工程将自己注册至discover-server。

导入：资料\\基础代码\\dtx 父工程，此工程自带了discover-server，discover-server基于Eureka实现。

## 2.4导入案例工程dtx-seata-demo ##

dtx-seata-demo是seata的测试工程，根据业务需求需要创建两个dtx-seata-demo工程。

（1）导入dtx-seata-demo

导入：资料\\基础代码\\dtx-seata-demo到父工程dtx下。

两个测试工程如下：

dtx/dtx-seata-demo/dtx-seata-demo-bank1 ，操作张三账户，连接数据库bank1  
       dtx/dtx-seata-demo/dtx-seata-demo-bank2 ，操作李四账户，连接数据库bank2

（2）父工程maven依赖说明

在dtx父工程中指定了SpringBoot和SpringCloud版本

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring‐boot‐dependencies</artifactId>
        <version>2.1.3.RELEASE</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring‐cloud‐dependencies</artifactId>
        <version>Greenwich.RELEASE</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>

在dtx-seata-demo父工程中指定了spring-cloud-alibaba-dependencies的版本。

<dependency>
        <groupId>com.alibaba.cloud</groupId>
        <artifactId>spring‐cloud‐alibaba‐dependencies</artifactId>
        <version>2.1.0.RELEASE</version>
        <type>pom</type>
        <scope>import</scope>
    </dependency>

（3）配置seata

在src/main/resource中，新增registry.conf、file.conf文件，内容可拷贝seata-server-0.7.1中的配置文件子。在registry.conf中registry.type使用file：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 2][]

在file.conf中更改service.vgroup\_mapping.\[springcloud服务名\]-fescar-service-group = "default"，并修改service.default.grouplist =\[seata服务端地址\]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 3][]

关于vgroup\_mapping的配置：

*  vgroup\_mapping.事务分组服务名=Seata Server集群名称（默认名称为default）
 *  default.grouplist = Seata Server集群地址

在 org.springframework.cloud:spring-cloud-starter-alibaba-seata 的org.springframework.cloud.alibaba.seata.GlobalTransactionAutoConfiguration 类中，默认会使用$\{spring.application.name\}-fescar-service-group 作为事务分组服务名注册到 Seata Server上，如果和file.conf 中的配置不一致，会提示 no available server to connect 错误,也可以通过配置 spring.cloud.alibaba.seata.tx-service-group 修改后缀，但是必须和 file.conf 中的配置保持一致。

（4）创建代理数据源

新增DatabaseConfiguration.java，Seata的RM通过DataSourceProxy才能在业务代码的事务提交时，通过这个切入点，与TC进行通信交互、记录undo\_log等。\\

package com.mqc.seatademo.config;
    
    import com.alibaba.druid.pool.DruidDataSource;
    import io.seata.rm.datasource.DataSourceProxy;
    import org.springframework.boot.context.properties.ConfigurationProperties;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    import org.springframework.context.annotation.Primary;
    
    import javax.sql.DataSource;
    
    /**
     * 数据源配置
     */
    @Configuration
    public class DatabaseConfiguration {
    
        @Bean
        @ConfigurationProperties("spring.datasource.ds0")
        public DruidDataSource ds0(){
            DruidDataSource druidDataSource =  new DruidDataSource();
            return druidDataSource;
        }
    
        /**
         * seata框架代理数据源
         * @param ds0
         * @return
         */
        @Primary
        @Bean
        public DataSource dataSource(DruidDataSource ds0)  {
            DataSourceProxy pds0 = new DataSourceProxy(ds0);
            return pds0;
        }
    }

## 2.5Seata的执行流程 ##

### 2.5.1 正常提交流程 ###

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 4][]

### 2.5.2回滚流程 ###

回滚流程省略前的RM注册过程。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 5][]

要点说明：

1、每个RM使用DataSourceProxy连接数据库，其目的是使用ConnectionProxy，使用数据源和数据连接代理的目的就是在第一阶段将undo\_log和业务数据放在一个本地事务提交，这样就保存了只要有业务操作就一定有undo\_log。

2、在第一阶段undo\_log中存放了数据修改前和修改后的值，为事务回滚作好准备，所以第一阶段完成就已经将分支事务提交，也就释放了锁资源。

3、TM开启全局事务开始，将XID全局事务id放在事务上下文中，通过feign调用也将XID传入下游分支事务，每个分支事务将自己的Branch ID分支事务ID与XID关联。

4、第二阶段全局事务提交，TC会通知各各分支参与者提交分支事务，在第一阶段就已经提交了分支事务，这里各各参与者只需要删除undo\_log即可，并且可以异步执行，第二阶段很快可以完成。

5、第二阶段全局事务回滚，TC会通知各各分支参与者回滚分支事务，通过 XID 和 Branch ID 找到相应的回滚日志，通过回滚日志生成反向的 SQL 并执行，以完成分支事务回滚到之前的状态，如果回滚失败则会重试回滚操作。

## 2.6dtx-seata-demo-bank1 ##

dtx-seata-demo-bank1实现如下功能：1、张三账户减少金额，开启全局事务。2、远程调用bank2向李四转账。

（1）DAO

package com.mqc.seatademo.dao;
    
    import org.apache.ibatis.annotations.Mapper;
    import org.apache.ibatis.annotations.Param;
    import org.apache.ibatis.annotations.Update;
    import org.springframework.stereotype.Component;
    
    @Mapper
    @Component
    public interface AccountDao {
        //更新账户金额
        @Update("update account_info set account_balance = account_balance + #{amount} where account_no = #{accountNo}")
        int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);
    }

（2）FeignClient

远程调用bank2的客户端

package com.mqc.seatademo.spring;
    
    import org.springframework.cloud.openfeign.FeignClient;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RequestParam;
    
    @FeignClient(value="seata-demo-bank2",fallback=Bank2ClientFallback.class)
    public interface Bank2Client {
        /**
         * 远程调用李四的微服务
         * @param amount 对应需要扣减的金额
         */
        @GetMapping("/bank2/transfer")
        String transfer(@RequestParam("amount") Double amount);
    }

（3）Service

package com.mqc.seatademo.service;
    
    import com.mqc.seatademo.dao.AccountDao;
    import com.mqc.seatademo.spring.Bank2Client;
    import io.seata.core.context.RootContext;
    import io.seata.spring.annotation.GlobalTransactional;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    
    @Service
    @Slf4j
    public class AccountInfoServiceImpl implements IAccountInfoService {
        @Autowired
        private AccountDao accountDao;
        @Autowired
        private Bank2Client bank2Client;
    
        /**
         * 扣减金额
         * @param accountNo
         * @param amount
         */
        @Override
        @Transactional
        @GlobalTransactional // 开启全局事务
        public void updateAccountBalance(String accountNo, Double amount) {
            log.info("bank1 service begin,XID：{}", RootContext.getXID());
            accountDao.updateAccountBalance(accountNo,amount*-1);
    
            // 远程调用bank2服务进行增加金额
            String transfer = bank2Client.transfer(amount);
            if("fallback".equals(transfer)){
                //调用李四微服务异常
                throw new RuntimeException("调用李四微服务异常");
            }
    
            if(amount == 2){
                //人为制造异常,调试使用
                throw new RuntimeException("bank1 make exception..");
            }
        }
    }

将@GlobalTransactional注解标注在全局事务发起的Service实现方法上，开启全局事务：GlobalTransactionalInterceptor会拦截@GlobalTransactional注解的方法，生成全局事务ID(XID)，XID会在整个  
分布式事务中传递。在远程调用时，spring-cloud-alibaba-seata会拦截Feign调用将XID传递到下游服务。

（6）Controller

package com.mqc.seatademo.controller;
    
    import com.mqc.seatademo.service.IAccountInfoService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.PathVariable;
    import org.springframework.web.bind.annotation.RequestParam;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @author Administrator
     * @version 1.0
     **/
    @RestController
    public class Bank1Controller {
    
        @Autowired
        IAccountInfoService accountInfoService;
    
        //张三转账
        @GetMapping("/transfer")
        public String transfer(@RequestParam("amount") Double amount){
            accountInfoService.updateAccountBalance("1",amount);
            return "bank1"+amount;
        }
    }

## 2.7dtx-seata-demo-bank2 ##

dtx-seata-demo-bank2实现如下功能：1、李四账户增加金额。dtx-seata-demo-bank2在本账号事务中作为分支事务不使用@GlobalTransactional。

（1）DAO

package com.mqc.seatademo.bank2.dao;
    
    import org.apache.ibatis.annotations.Mapper;
    import org.apache.ibatis.annotations.Param;
    import org.apache.ibatis.annotations.Update;
    import org.springframework.stereotype.Component;
    
    /**
     * Created by Administrator.
     */
    @Mapper
    @Component
    public interface AccountInfoDao {
        //更新账户
        @Update("UPDATE account_info SET account_balance = account_balance + #{amount} WHERE account_no = #{accountNo}")
        int updateAccountBalance(@Param("accountNo") String accountNo, @Param("amount") Double amount);
    
    }

（2）Service

package com.mqc.seatademo.bank2.service.impl;
    
    import com.mqc.seatademo.bank2.dao.AccountInfoDao;
    import com.mqc.seatademo.bank2.service.AccountInfoService;
    import io.seata.core.context.RootContext;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.transaction.annotation.Transactional;
    
    /**
     * @author Administrator
     * @version 1.0
     **/
    @Service
    @Slf4j
    public class AccountInfoServiceImpl implements AccountInfoService {
    
        @Autowired
        AccountInfoDao accountInfoDao;
    
    
        @Transactional
        @Override
        public void updateAccountBalance(String accountNo, Double amount) {
            log.info("bank2 service begin,XID：{}",RootContext.getXID());
            //李四增加金额
            accountInfoDao.updateAccountBalance(accountNo,amount);
            if(amount==3){
                //人为制造异常
                throw new RuntimeException("bank2 make exception..");
            }
        }
    }

（3）Controller

package com.mqc.seatademo.bank2.controller;
    
    import com.mqc.seatademo.bank2.service.AccountInfoService;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    /**
     * @author Administrator
     * @version 1.0
     **/
    @RestController
    public class Bank2Controller {
    
        @Autowired
        AccountInfoService accountInfoService;
    
        //接收张三的转账
        @GetMapping("/transfer")
        public String transfer(Double amount){
            //李四增加金额
            accountInfoService.updateAccountBalance("2",amount);
            return "bank2"+amount;
        }
    }

## 2.8测试场景 ##

*  张三向李四转账成功。
 *  李四事务失败，张三事务回滚成功。
 *  张三事务失败，李四事务回滚成功。
 *  分支事务超时测试

## 2.9小结 ##

本节讲解了传统2PC（基于数据库XA协议）和Seata实现2PC的两种2PC方案，由于Seata的0侵入性并且解决了传统2PC长期锁资源的问题，所以推荐采用Seata实现2PC。

Seata实现2PC要点：

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

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

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

好了，这个是我目前学习的seata框架解决分布式事务的问题，后面我可能还会进一步去学习seata框架的源码，有需要的可以关注我的博客，我后面也会出对应的seata源码解析学习分享博客纪录。上面的代码跟资料的github地址是：[https://github.com/JokerMqc/seata-demo.git][https_github.com_JokerMqc_seata-demo.git]，大家有需要可以拷贝下来学习。

[20210527162930321.png]: /images/20221014/95dae6bed5cd4b2c8cbf1c5d6dc10abc.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70]: /images/20221014/669c0c86dde54f0c8858267752817bbb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 1]: /images/20221014/172f6ff653e74770a27daac1ff4bc2f9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 2]: /images/20221014/cd23a0f3ef694969b6066613dedf2b5a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 3]: /images/20221014/41c4b848680f48f4a9e3c68bf1692332.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 4]: /images/20221014/c9be2694e0f8473b8e38a5e29702bcec.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2pva2VNcWM_size_16_color_FFFFFF_t_70 5]: /images/20221014/543078589c94403caa4eed7a1b46ec5e.png
[https_github.com_JokerMqc_seata-demo.git]: https://github.com/JokerMqc/seata-demo.git