[Java知识库]Springboot集成分布式事务Seata

概述

springboot+springcloud+seata

版本选择

springboot：2.1.3.RELEASE；
springcloud：Greenwich.RELEASE
alibaba-seata:2.1.0.RELEASE

模块组成

父模块+子模块

pom.xml文件

父模块

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <groupId>cn.itcast.dtx</groupId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <artifactId>dtx-seata-demo</artifactId>
    <modules>
        <module>dtx-seata-demo-bank1</module>
        <module>dtx-seata-demo-bank2</module>
    </modules>
    <properties>
        <spring-cloud-alibaba.version>2.1.0.RELEASE</spring-cloud-alibaba.version>
    </properties>
    <!-- 只是用来管理包的版本，并不会导入依赖包，需要导入哪些依赖包是由子模块声明的-->
    <dependencyManagement>
        <dependencies>
         <!-- 下面的三个Pom类型的依赖必须，负责版本管理，否则依赖包导入失败-->
            <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>
            <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>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger2</artifactId>
                <version>2.9.2</version>
            </dependency>
            <dependency>
                <groupId>io.springfox</groupId>
                <artifactId>springfox-swagger-ui</artifactId>
                <version>2.9.2</version>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.0</version>
            </dependency>
            <dependency>
                <groupId>javax.servlet</groupId>
                <artifactId>javax.servlet-api</artifactId>
                <version>3.1.0</version>
                <scope>provided</scope>
            </dependency>
            <dependency>
                <groupId>javax.interceptor</groupId>
                <artifactId>javax.interceptor-api</artifactId>
                <version>1.2</version>
            </dependency>
            <dependency>
                <groupId>mysql</groupId>
                <artifactId>mysql-connector-java</artifactId>
                <version>5.1.47</version>
            </dependency>
            <dependency>
                <groupId>org.mybatis.spring.boot</groupId>
                <artifactId>mybatis-spring-boot-starter</artifactId>
                <version>2.0.0</version>
            </dependency>
            <dependency>
                <groupId>com.alibaba</groupId>
                <artifactId>druid-spring-boot-starter</artifactId>
                <version>1.2.3</version>
            </dependency>
            <dependency>
                <groupId>commons-lang</groupId>
                <artifactId>commons-lang</artifactId>
                <version>2.6</version>
            </dependency>
        </dependencies>
    </dependencyManagement>
</project>

子模块

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>dtx-seata-demo</artifactId>
        <groupId>cn.itcast.dtx</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>
    <artifactId>dtx-seata-demo-bank1</artifactId>
    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-seata</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-starter-openfeign</artifactId>
        </dependency>
        <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>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
        </dependency>
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
        </dependency>
    </dependencies>
</project>

配置

配置如下图：

spring配置

application.yml

spring:
  application:
    name: seata-demo-bank1
  profiles: 
    active: dev
  main:
    allow-bean-definition-overriding: true
  jackson:
    date-format: yyyy-MM-dd HH:mm:ss
    time-zone: GMT+8
server:
  servlet:
    context-path: /bank1
feign:
  hystrix:
    enabled: true
  compression:
    request:
      enabled: true # 配置请求GZIP压缩
      mime-types: ["text/xml","application/xml","application/json"] # 配置压缩支持的MIME TYPE
      min-request-size: 2048 # 配置压缩数据大小的下限
    response:
      enabled: true # 配置响应GZIP压缩

上面的配置指定的了dev配置环境，所以这里使用application-dev.yml命名
application-dev.yml

server: 
  port: 56081
spring:
  ##################### DB #####################
  datasource:
    druid:
      url: jdbc:mysql://127.0.0.1:3306/practice?useSSL=false&serverTimezone=Asia/Shanghai
      username: root
      password: 123456
      driver-class-name: com.mysql.jdbc.Driver
      initialSize: 5
      minIdle: 5
      maxActive: 20
      maxWait: 60000
      timeBetweenEvictionRunsMillis: 60000
      minEvictableIdleTimeMillis: 300000
      validationQuery: SELECT user()
      testWhileIdle: true
      testOnBorrow: false
      testOnReturn: false
      poolPreparedStatements: true
      connection-properties: druid.stat.mergeSql:true;druid.stat.slowSqlMillis:5000
logging:
  level:
    root: INFO
    io:
      seata: DEBUG
    org:
      springframework:
        cloud:
          alibaba:
            seata:
              web: DEBUG

seata客户端配置

我们这里使用最简单的file类型来连接seata服务器，所以配置也是要file类型。

registry.conf

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  # 这里使用file类型连接服务器
  type = "file"
  # 指定file的 配置文件
  file {
    name = "file.conf"
  }
}

config {
  # file、nacos 、apollo、zk
  # 指定file类型
  type = "file"

  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
  }
}
## transaction log store
store {
  ## store mode: file、db
  mode = "file"

  ## file store
  file {
    dir = "sessionStore"

    # 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
    # async, sync
    flush-disk-mode = async
  }
}

service {
  #vgroup->rgroup
  # 命名规则是固定的：vgroup_mapping.[springcloud服务名]-fescar-service-group
  # 源码根据这个命名规则获取应用服务的信息
  vgroup_mapping.seata-demo-bank1-fescar-service-group = "default"
  #only support single node，指定seata server的地址
  default.grouplist = "127.0.0.1:8888"
  #degrade current not support
  enableDegrade = false
  #disable
  disable = false
}
client {
  async.commit.buffer.limit = 10000
  lock {
    retry.internal = 10
    retry.times = 30
  }
}

配置seata代理数据源

配置位置如下：

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

package cn.itcast.dtx.seatademo.bank1.config;

import com.alibaba.druid.pool.DruidDataSource;
import io.seata.rm.datasource.DataSourceProxy;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.ApplicationContext;
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 {
    private final ApplicationContext applicationContext;

    public DatabaseConfiguration(ApplicationContext applicationContext) {
        this.applicationContext = applicationContext;
    }

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource.druid")
    public DruidDataSource ds0() {
        DruidDataSource druidDataSource = new DruidDataSource();
        return druidDataSource;
    }

    @Primary
    @Bean
    public DataSource dataSource(DruidDataSource ds0)  {
        DataSourceProxy pds0 = new DataSourceProxy(ds0);
        return pds0;
    }
}

启动类修改

注意要把springboot自带的数据源带来排除掉，否则出现配置的代理数据源与springboot自带的形成循环依赖。

package cn.itcast.dtx.seatademo.bank1;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
import org.springframework.cloud.openfeign.EnableFeignClients;
// 启动是排除springboot自带的数据源配置类
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
@EnableFeignClients(basePackages = {"cn.itcast.dtx.seatademo.bank1.spring"})
public class Bank1Server {
    public static void main(String[] args) {
        SpringApplication.run(Bank1Server.class, args);
    }
}

添加undo_log表

该表用来事务回滚，分支事务提交时记录事务相关信息，在分布式事务异常时回滚，分布式事务结束后会删除undo_log的记录。
在spring配置指定的数据库中创建表，每个需要注册到seata server的业务模块都有创建该表，创建语句如下：

-- 注意此处0.3.0+ 增加唯一索引 ux_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;

seata server服务

seata分成服务端和客户端，客户端我们在项目中引入alibaba-seata即可，服务端需要单独部署。
这里只介绍最简单的单独部署，启动模式file。
这里选择版本：seata-server-1.2.0
下载地址：https://github.com/seata/seata/tags

启动

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

8888为服务对外服务端口，这里需要和file.conf中指定的一致；-m file表示以file模式启动。

seata使用示例

这里只简单的演示在一个微服务中，一个分布式事物包含两个分支事物。

创建业务表

CREATE TABLE `t_order` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `order_id` varchar(255) DEFAULT NULL COMMENT '订单',
  `create_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

CREATE TABLE `t_user` (
  `id` int(11) NOT NULL AUTO_INCREMENT COMMENT '主键',
  `user_name` varchar(255) DEFAULT NULL COMMENT '用户名称',
  `sex` tinyint(1) DEFAULT '0' COMMENT '性别，0：男，1：女',
  `age` int(3) DEFAULT '0' COMMENT '年龄',
  `address` varchar(255) DEFAULT NULL COMMENT '地址',
  `phone` varchar(11) DEFAULT NULL COMMENT '手机',
  `create_time` datetime DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5 DEFAULT CHARSET=utf8;

事务测试类

代码结构如下：

controller

package cn.itcast.dtx.seatademo.bank1.controller;

import cn.itcast.dtx.seatademo.bank1.service.TestGlobalTransService;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;

/**
 * @Description :
 * @Version : V1.0.0
 * @Date : 2022/7/1 15:16
 */
@RestController
@RequestMapping(value = "/test/")
public class TestGlobalTrans {
    @Resource
    private TestGlobalTransService service;
    @GetMapping("/seataTrans")
    public String testSeataTrans() throws Exception {
        service.testTrans();
        return "success";
    }
}

service

package cn.itcast.dtx.seatademo.bank1.service;

import io.seata.spring.annotation.GlobalTransactional;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;

/**
 * @Description :
 * @Version : V1.0.0
 * @Date : 2022/7/1 15:18
 */
@Service
public class TestGlobalTransService {

    @Resource
    private TestOnlyTransService service;
	// 只需要添加@GlobalTransactional注解就说明启动了分布式事务
    @GlobalTransactional
    public void testTrans() throws Exception {
    	// 分支事务添加用户信息
        service.insertUser();
        // 分支事务添加订单信息
        service.insertOrder();
        // 抛出异常，事务回滚
        throw new Exception("test exception");
    }
}

负责分支事务的服务类，单独拿出来是因为spring事务代理要求事务方法如果和调用方法放在一个类中，代理不生效，具体原因不在赘述。

package cn.itcast.dtx.seatademo.bank1.service;

import cn.itcast.dtx.seatademo.bank1.dao.UserInfo;
import cn.itcast.dtx.seatademo.bank1.dao.OrderInfo;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import javax.annotation.Resource;

/**
 * @Description :
 * @Author : 
 * @Version : V1.0.0
 * @Date : 2022/7/1 15:18
 */
@Service
public class TestOnlyTransService {

    @Resource
    private OrderInfo orderInfo;

    @Resource
    private UserInfo userInfo;

    @Transactional(rollbackFor = Exception.class)
    public void insertOrder() {
        orderInfo.insert();
    }

    @Transactional(rollbackFor = Exception.class)
    public void insertUser() {
        userInfo.insert();
    }
}

dao

package cn.itcast.dtx.seatademo.bank1.dao;

import org.apache.ibatis.annotations.Insert;
import org.apache.ibatis.annotations.Mapper;
import org.springframework.stereotype.Component;

/**
 * Created by Administrator.
 */
@Mapper
@Component
public interface OrderInfo {

    @Insert("insert into t_order (order_id,create_time) values ('aaaaaaaaa',now())")
    int insert();
}

@Mapper
@Component
public interface UserInfo {

    @Insert("insert into t_user (user_name,sex,age, create_time) values ('aaaaaaaaa', 0,18,now())")
    int insert();

}

测试结果

1. 浏览器调用：http://127.0.0.1:56081/bank1/test/seataTrans
2. 因为 我们的代码中有异常抛出，所有数据库中没有插入数据，事务回滚了。

2022-07-03 11:31:41.697  INFO 13816 --- [atch_RMROLE_1_8] io.seata.rm.AbstractRMHandler            : Branch Rollbacked result: PhaseTwo_Rollbacked
2022-07-03 11:31:41.697 DEBUG 13816 --- [atch_RMROLE_1_8] i.s.core.rpc.netty.RmMessageListener     : branch rollback result:xid=169.254.144.154:8888:2015847663,branchId=2015847665,branchStatus=PhaseTwo_Rollbacked,result code =Success,getMsg =null
2022-07-03 11:31:41.697 DEBUG 13816 --- [atch_RMROLE_1_8] i.s.core.rpc.netty.AbstractRpcRemoting   : send response:xid=169.254.144.154:8888:2015847663,branchId=2015847665,branchStatus=PhaseTwo_Rollbacked,result code =Success,getMsg =null,channel:[id: 0xbf86ca50, L:/127.0.0.1:62389 - R:/127.0.0.1:8888]
2022-07-03 11:31:41.701 DEBUG 13816 --- [lector_RMROLE_1] i.s.core.rpc.netty.AbstractRpcRemoting   : io.seata.core.rpc.netty.RmRpcClient@6e6f2c4f msgId:3, body:xid=169.254.144.154:8888:2015847663,branchId=2015847664,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/practice,applicationData=null
2022-07-03 11:31:41.701  INFO 13816 --- [atch_RMROLE_2_8] i.s.core.rpc.netty.RmMessageListener     : onMessage:xid=169.254.144.154:8888:2015847663,branchId=2015847664,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/practice,applicationData=null
2022-07-03 11:31:41.702  INFO 13816 --- [atch_RMROLE_2_8] io.seata.rm.AbstractRMHandler            : Branch Rollbacking: 169.254.144.154:8888:2015847663 2015847664 jdbc:mysql://127.0.0.1:3306/practice
2022-07-03 11:31:41.708  INFO 13816 --- [atch_RMROLE_2_8] i.s.rm.datasource.undo.UndoLogManager    : xid 169.254.144.154:8888:2015847663 branch 2015847664, undo_log deleted with GlobalFinished
2022-07-03 11:31:41.709  INFO 13816 --- [atch_RMROLE_2_8] io.seata.rm.AbstractRMHandler            : Branch Rollbacked result: PhaseTwo_Rollbacked
2022-07-03 11:31:41.709 DEBUG 13816 --- [atch_RMROLE_2_8] i.s.core.rpc.netty.RmMessageListener     : branch rollback result:xid=169.254.144.154:8888:2015847663,branchId=2015847664,branchStatus=PhaseTwo_Rollbacked,result code =Success,getMsg =null
2022-07-03 11:31:41.709 DEBUG 13816 --- [atch_RMROLE_2_8] i.s.core.rpc.netty.AbstractRpcRemoting   : send response:xid=169.254.144.154:8888:2015847663,branchId=2015847664,branchStatus=PhaseTwo_Rollbacked,result code =Success,getMsg =null,channel:[id: 0xbf86ca50, L:/127.0.0.1:62389 - R:/127.0.0.1:8888]
2022-07-03 11:31:41.714 DEBUG 13816 --- [io-56081-exec-4] io.seata.core.context.RootContext        : unbind 169.254.144.154:8888:2015847663
2022-07-03 11:31:41.714  INFO 13816 --- [io-56081-exec-4] i.seata.tm.api.DefaultGlobalTransaction  : [169.254.144.154:8888:2015847663] rollback status:Rollbacked
2022-07-03 11:31:41.734 ERROR 13816 --- [io-56081-exec-4] o.a.c.c.C.[.[.[.[dispatcherServlet]      : Servlet.service() for servlet [dispatcherServlet] in context with path [/bank1] threw exception [Request processing failed; nested exception is java.lang.Exception: test exception] with root cause

java.lang.Exception: test exception
	at cn.itcast.dtx.seatademo.bank1.service.TestGlobalTransService.testTrans(TestGlobalTransService.java:24) ~[classes/:na]
	at cn.itcast.dtx.seatademo.bank1.service.TestGlobalTransService$$FastClassBySpringCGLIB$$dcfd65aa.invoke(<generated>) ~[classes/:na]
	at org.springframework.cglib.proxy.MethodProxy.invoke(MethodProxy.java:218) ~[spring-core-5.1.5.RELEASE.jar:5.1.5.RELEASE]
	at org.springframework.aop.framework.CglibAopProxy$CglibMethodInvocation.invokeJoinpoint(CglibAopProxy.java:749) ~[spring-aop-5.1.5.RELEASE.jar:5.1.5.RELEASE]

3. 修改代码，删除异常，数据保存成功。

  @GlobalTransactional
    public void testTrans() throws Exception {
        service.insertUser();
        service.insertOrder();
//        throw new Exception("test exception");
    }