Spring Boot 集成Seata(fescar)实现分布式事务处理以及实现机制

àì夳堔傛蜴生んèń 2021-11-05 13:30 249阅读 0赞

## **1.介绍** ##

2019 年 1 月，阿里巴巴中间件团队发起了开源项目 [*Fescar*][Fescar]*（Fast & EaSy Commit And Rollback）*，和社区一起共建开源分布式事务解决方案。Fescar 的愿景是让分布式事务的使用像本地事务的使用一样，简单和高效，并逐步解决开发者们遇到的分布式事务方面的所有难题。

**Fescar 开源后，蚂蚁金服加入 Fescar 社区参与共建，并在 Fescar 0.4.0 版本中贡献了 TCC 模式。**

为了打造更中立、更开放、生态更加丰富的分布式事务开源社区，经过社区核心成员的投票，大家决定对 Fescar 进行品牌升级，并更名为 **Seata**，意为：**Simple Extensible Autonomous Transaction Architecture**，是一套一站式分布式事务解决方案。

Seata 融合了阿里巴巴和蚂蚁金服在分布式事务技术上的积累，并沉淀了新零售、云计算和新金融等场景下丰富的实践经验，但要实现**适用于所有的分布式事务场景**的愿景，仍有很长的路要走。

Seata现在还不够稳定，在企业中可能大部分要求不够高的分布式情况下，还是不会处理事物或者使用MQ，TCC等来实现分布式事务，但是对于MQ而言来保证最终强一致性，会增加很多的代码，所以也希望Seata能代替他们，越来越成熟。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70][]

对于Seata源码感兴趣的可以参考文章[Seata源码解析][Seata]

## **2.案列** ##

**环境：**

*  SpringBoot 2.1.6.RELEASE
 *  Dubbo 2.7.1
 *  Mybatis 3.5.1
 *  Seata 0.7.1
 *  Zookeeper 3.4.10
 *  Mysql8

在此之前你可能需要下载Zookeeper和Seata ，需要注意的是项目中使用的Seata版本和安装版本最好一致，要不然会出现意想不到的错误。

**2.1、业务场景**

为了简化流程，我们只需要订单和库存两个服务。创建订单的时候，调用库存服务，扣减库存。

涉及的表设计如下：

CREATE TABLE `t_order` (
     `id` int(11) NOT NULL AUTO_INCREMENT,
     `order_no` varchar(255) DEFAULT NULL,
     `user_id` varchar(255) DEFAULT NULL,
     `commodity_code` varchar(255) DEFAULT NULL,
     `count` int(11) DEFAULT '0',
     `amount` double(14,2) DEFAULT '0.00',
     PRIMARY KEY (`id`)
    ) ENGINE=InnoDB AUTO_INCREMENT=38 DEFAULT CHARSET=utf8;
     
    CREATE TABLE `t_storage` (
     `id` int(11) NOT NULL AUTO_INCREMENT,
     `commodity_code` varchar(255) DEFAULT NULL,
     `name` varchar(255) DEFAULT NULL,
     `count` int(11) DEFAULT '0',
     PRIMARY KEY (`id`),
     UNIQUE KEY `commodity_code` (`commodity_code`)
    ) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8;

**Seata的下载安装**

打开[https://github.com/seata/seata/releases][https_github.com_seata_seata_releases]，我下的新版本是v0.7.1。

下载解压后，到seata-server-0.7.1\\distribution\\bin目录下可以看到seata-server.bat和seata-server.sh，选择一个双击执行。  
不出意外的话，当你看到-Server started ...等字样，就正常启动了。

下面说一说项目中的主要代码以及Seata的工作原理：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 1][]

这是项目的整体架构，common是使用Dubbo暴露出来的RMI接口，order是订单服务，Storage是库存服务。

下面是项目中的主要依赖：

<dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
            </dependency>
            <dependency>
                <groupId>org.mybatis.spring.boot</groupId>
                <artifactId>mybatis-spring-boot-starter</artifactId>
                <version>2.0.1</version>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>8.0.12</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo</artifactId>
                <version>2.7.1</version>
            </dependency>
            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo-spring-boot-starter</artifactId>
                <version>2.7.1</version>
            </dependency>
            
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-framework</artifactId>
                <version>2.13.0</version>
            </dependency>
            <dependency>
                <groupId>org.apache.curator</groupId>
                <artifactId>curator-recipes</artifactId>
                <version>2.13.0</version>
            </dependency>
            <dependency>
                <groupId>io.seata</groupId>
                <artifactId>seata-all</artifactId>
                <version>0.7.1</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-test</artifactId>
                <scope>test</scope>
            </dependency>
        </dependencies>

项目配置：

order服务的application.properties

server.port=8011
    
    spring.application.name=springboot-order
    spring.http.encoding.charset=UTF-8
    spring.http.encoding.enabled=true
    spring.http.encoding.force=true
    spring.datasource.url=jdbc:mysql://127.0.0.1:3306/seata?useSSL=false&useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=UTC
    spring.datasource.username=root
    spring.datasource.password=123456
    
    
    mybatis.mapper-locations=classpath*:/mapper/*.xml
    mybatis.type-aliases-package=com.viewscenes.order.entity
    logging.level.com.viewscenes.order.mapper=debug
    
    dubbo.application.name=order-service
    dubbo.registry.address=zookeeper://127.0.0.1:2181
    dubbo.protocol.name=dubbo
    dubbo.protocol.port=20881
    dubbo.consumer.timeout=9999999
    dubbo.consumer.check=false

storage服务的：

server.port=8012
    
    spring.application.name=springboot-storage
    spring.http.encoding.charset=UTF-8
    spring.http.encoding.enabled=true
    spring.http.encoding.force=true
    spring.datasource.url=jdbc:mysql://127.0.0.1:3306/seata?useSSL=false&useUnicode=true&characterEncoding=utf-8&allowMultiQueries=true&serverTimezone=UTC
    spring.datasource.username=root
    spring.datasource.password=123456
    
    
    mybatis.mapper-locations=classpath*:/mapper/*.xml
    mybatis.type-aliases-package=com.viewscenes.storage.entity
    logging.level.com.viewscenes.storage.mapper=debug
    
    dubbo.application.name=storage-service
    dubbo.registry.address=zookeeper://127.0.0.1:2181
    dubbo.protocol.name=dubbo
    dubbo.protocol.port=20882
    dubbo.consumer.timeout=9999999
    dubbo.consumer.check=false

**数据源**

Seata 是通过代理数据源实现事务分支，所以需要先配置一个数据源的代理，否则事务不会回滚。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 2][]

注意，这里的DataSourceProxy类位于io.seata.rm.datasource包内。

**Seata配置**

还需要配置全局事务扫描器。有两个参数，一个是应用名称，一个是事务分组。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 3][]

关于Seata事务的一系列初始化工作都在这里完成。

**配置注册中心**

Seata连接到服务器的时候需要一些配置项，这时候有一个registry.conf文件可以指定注册中心和配置文件是什么。

这里有很多可选性，比如file、nacos 、apollo、zk、consul。

后面4个都是业界成熟的配置注册中心产品，为啥还有个file呢？

官方的初衷是在不依赖第三方配置注册中心的基础上快速集成测试seata功能，但是file类型本身不具备注册中心的动态发现和动态配置功能。

本次选用的是默认file的注册，如果你想使用其他注册中心，需要更改seata的下载安装的配置文件和项目中的Regis文件，下面是file文件的配置

registry.conf

registry {
    type = "file"
    
    file {
      name = "file.conf"
    }
    }
    config {
      # file、nacos 、apollo、zk、consul
      type = "file"
    
      file {
        name = "file.conf"
      }
    }

如果你选择了file类型，通过name属性指定了file.conf，这个文件中指定了客户端或服务器的配置信息。比如传输协议、服务器地址等。

file.conf

transport {
      # tcp udt unix-domain-socket
      type = "TCP"
      #NIO NATIVE
      server = "NIO"
      #enable heartbeat
      heartbeat = true
      #thread factory for netty
      thread-factory {
        boss-thread-prefix = "NettyBoss"
        worker-thread-prefix = "NettyServerNIOWorker"
        server-executor-thread-prefix = "NettyServerBizHandler"
        share-boss-worker = false
        client-selector-thread-prefix = "NettyClientSelector"
        client-selector-thread-size = 1
        client-worker-thread-prefix = "NettyClientWorkerThread"
        # netty boss thread size,will not be used for UDT
        boss-thread-size = 1
        #auto default pin or 8
        worker-thread-size = 8
      }
    }
    
    service {
      #vgroup->rgroup
      vgroup_mapping.my_test_tx_group = "default"
      #only support single node
      default.grouplist = "127.0.0.1:8091"
      #degrade current not support
      enableDegrade = false
      #disable
      disable = false
    }
    
    client {
      async.commit.buffer.limit = 10000
      lock {
        retry.internal = 10
        retry.times = 30
      }
    }
    ## transaction log store
    store {
      ## store mode: file、db
      mode = "file"
    
      ## file store
      file {
        dir = "file_store/data"
    
        # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
        max-branch-session-size = 16384
        # globe session size , if exceeded throws exceptions
        max-global-session-size = 512
        # file buffer size , if exceeded allocate new buffer
        file-write-buffer-cache-size = 16384
        # when recover batch read size
        session.reload.read_size = 100
      }
    
      ## database store
      db {
        driver_class = ""
        url = ""
        user = ""
        password = ""
      }
    }

业务代码：

库存中主要是更改库存业务：

![20190807100021643.png][]

订单业务：

在订单服务中，先扣减库存，再创建订单。最后抛出异常，然后去数据库检查事务是否回滚。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 4][]

在订单服务事务开始的方法上，需要标注@GlobalTransactional。另外，在库存服务的方法里，不需要此注解，事务会通过Dubbo进行传播。

到这就差不多了，使用PostMan进行测试下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 5][]

使用断点测试然后随时查看数据库变化会更好的看出事务的回滚机制。

下面是项目代码地址：[项目地址][Link 1]

**Seata的实现机制**

我们发现，我们就仅仅增加了seata 的`@GlobalTransactional`注解，就实现了分布式事务。其实 seata 增加了个拦截器来专门处理被`@GlobalTransactional`注解的方法，即`GlobalTransactionalInterceptor`，其分布式事务的执行流程都在这里完成的：

/**
     * The type Global transactional interceptor.
     */
    public class GlobalTransactionalInterceptor implements MethodInterceptor {
    
        private static final Logger LOGGER = LoggerFactory.getLogger(GlobalTransactionalInterceptor.class);
        private static final FailureHandler DEFAULT_FAIL_HANDLER = new DefaultFailureHandlerImpl();
    
        private final TransactionalTemplate transactionalTemplate = new TransactionalTemplate();
        private final GlobalLockTemplate<Object> globalLockTemplate = new GlobalLockTemplate<>();
        private final FailureHandler failureHandler;
    
        /**
         * Instantiates a new Global transactional interceptor.
         *
         * @param failureHandler the failure handler
         */
        public GlobalTransactionalInterceptor(FailureHandler failureHandler) {
            if (null == failureHandler) {
                failureHandler = DEFAULT_FAIL_HANDLER;
            }
            this.failureHandler = failureHandler;
        }
    
        @Override
        public Object invoke(final MethodInvocation methodInvocation) throws Throwable {
            Class<?> targetClass = (methodInvocation.getThis() != null ? AopUtils.getTargetClass(methodInvocation.getThis()) : null);
            Method specificMethod = ClassUtils.getMostSpecificMethod(methodInvocation.getMethod(), targetClass);
            final Method method = BridgeMethodResolver.findBridgedMethod(specificMethod);
    
            final GlobalTransactional globalTransactionalAnnotation = getAnnotation(method, GlobalTransactional.class);
            final GlobalLock globalLockAnnotation = getAnnotation(method, GlobalLock.class);
            if (globalTransactionalAnnotation != null) {
                return handleGlobalTransaction(methodInvocation, globalTransactionalAnnotation);
            } else if (globalLockAnnotation != null) {
                return handleGlobalLock(methodInvocation);
            } else {
                return methodInvocation.proceed();
            }
        }
    
        private Object handleGlobalLock(final MethodInvocation methodInvocation) throws Exception {
            return globalLockTemplate.execute(new Callable<Object>() {
                @Override
                public Object call() throws Exception {
                    try {
                        return methodInvocation.proceed();
                    } catch (Throwable e) {
                        if (e instanceof Exception) {
                            throw (Exception)e;
                        } else {
                            throw new RuntimeException(e);
                        }
                    }
                }
            });
        }
    
        private Object handleGlobalTransaction(final MethodInvocation methodInvocation,
                                               final GlobalTransactional globalTrxAnno) throws Throwable {
            try {
                return transactionalTemplate.execute(new TransactionalExecutor() {
                    @Override
                    public Object execute() throws Throwable {
                        return methodInvocation.proceed();
                    }
    
                    @Override
                    public int timeout() {
                        return globalTrxAnno.timeoutMills();
                    }
    
                    @Override
                    public String name() {
                        String name = globalTrxAnno.name();
                        if (!StringUtils.isNullOrEmpty(name)) {
                            return name;
                        }
                        return formatMethod(methodInvocation.getMethod());
                    }
                });
            } catch (TransactionalExecutor.ExecutionException e) {
                TransactionalExecutor.Code code = e.getCode();
                switch (code) {
                    case RollbackDone:
                        throw e.getOriginalException();
                    case BeginFailure:
                        failureHandler.onBeginFailure(e.getTransaction(), e.getCause());
                        throw e.getCause();
                    case CommitFailure:
                        failureHandler.onCommitFailure(e.getTransaction(), e.getCause());
                        throw e.getCause();
                    case RollbackFailure:
                        failureHandler.onRollbackFailure(e.getTransaction(), e.getCause());
                        throw e.getCause();
                    default:
                        throw new ShouldNeverHappenException("Unknown TransactionalExecutor.Code: " + code);
    
                }
            }
        }
    
        private <T extends Annotation> T getAnnotation(Method method, Class<T> clazz) {
            if (method == null) {
                return null;
            }
            return method.getAnnotation(clazz);
        }
    
        private String formatMethod(Method method) {
            String paramTypes = Arrays.stream(method.getParameterTypes())
                    .map(Class::getName)
                    .reduce((p1, p2) -> String.format("%s, %s", p1, p2))
                    .orElse("");
            return method.getName() + "(" + paramTypes + ")";
        }
    }

主要逻辑是在`TransactionalTemplate#execute`方法：

public Object execute(TransactionalExecutor business) throws TransactionalExecutor.ExecutionException {
    
            // 1. get or create a transaction
            GlobalTransaction tx = GlobalTransactionContext.getCurrentOrCreate();
    
            try {
    
                // 2. begin transaction
                try {
                    triggerBeforeBegin();
                    tx.begin(business.timeout(), business.name());
                    triggerAfterBegin();
                } catch (TransactionException txe) {
                    throw new TransactionalExecutor.ExecutionException(tx, txe,
                        TransactionalExecutor.Code.BeginFailure);
    
                }
                Object rs = null;
                try {
    
                    // Do Your Business
                    rs = business.execute();
    
                } catch (Throwable ex) {
    
                    // 3. any business exception, rollback.
                    try {
                        triggerBeforeRollback();
                        tx.rollback();
                        triggerAfterRollback();
                        // 3.1 Successfully rolled back
                        throw new TransactionalExecutor.ExecutionException(tx, TransactionalExecutor.Code.RollbackDone, ex);
    
                    } catch (TransactionException txe) {
                        // 3.2 Failed to rollback
                        throw new TransactionalExecutor.ExecutionException(tx, txe,
                            TransactionalExecutor.Code.RollbackFailure, ex);
    
                    }
    
                }
                // 4. everything is fine, commit.
                try {
                    triggerBeforeCommit();
                    tx.commit();
                    triggerAfterCommit();
                } catch (TransactionException txe) {
                    // 4.1 Failed to commit
                    throw new TransactionalExecutor.ExecutionException(tx, txe,
                        TransactionalExecutor.Code.CommitFailure);
                }
    
                return rs;
            } finally {
                //5. clear
                triggerAfterCompletion();
                cleanUp();
            }
        }

大致的执行流程为：

1.  获取全局事务信息：先从`ThreadLocal`中获取，如果没有则创建一个`DefaultGlobalTransaction`。
2.  开启全局事务`tx.begin(business.timeout(), business.name())`：通过`DefaultTransactionManager`的 begin 方法开启全局事务。`DefaultTransactionManager`负责 TM 与 TC 通讯，发送begin、commit、rollback指令。TC 接收到 TM 发过来的 begin 指令后，会返回一个全局唯一的 XID 给 TM。
3.  执行业务代码`business.execute()`：在每个本地事务中，会生成分支事务标识 BranchId, 然后根据业务 SQL 执行前后的镜像，生成 undoLog，并随着业务 SQL 一起提交。
4.  全局事务回滚`tx.rollback()`：当业务代码执行过程中抛出任何异常，都会进行全局事务的回滚操作。根据 XID 和 BranchId 查找 undoLog，然后反向生成业务 SQL，接着执行该 SQL，并且删除 undoLog 记录。
5.  全局事务提交`tx.commit()`：当业务代码执行正常时，则会提交全局事务。分支事务此时已经完成提交，只需要删除 undoLog 即可。

[Fescar]: https://www.oschina.net/p/fescar
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70]: /images/20211105/2489b4b014234c7dac62c39cf35a5aad.png
[Seata]: http://www.iocoder.cn/Seata/build-debugging-environment/
[https_github.com_seata_seata_releases]: https://github.com/seata/seata/releases
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 1]: /images/20211105/e5d9d690cc234f0bb16acdac092cf79d.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 2]: /images/20211105/15b24d54ec2a4450b25b9fc5c6921e94.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 3]: /images/20211105/a8fc023aa1fa44e682ff6ccc513444e8.png
[20190807100021643.png]: /images/20211105/55bcc8a8edc147c0abc06ba7c1debee9.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 4]: /images/20211105/ebaf4db148cb401e8c93ac66e3ad9e07.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM2ODI3OTU3_size_16_color_FFFFFF_t_70 5]: /images/20211105/394ea14e3da549c183be291cc14ea914.png
[Link 1]: https://download.csdn.net/download/qq_36827957/11489461