二、springboot項目使用seata實現分布式事務

所有文章

https://www.cnblogs.com/lay2017/p/12078232.html

 

正文

在上一篇文章中，我們簡單地了解了一下什么是seata。它是來自阿里巴巴的內部項目不斷地發展出來的。2019年以fescar命名開源於apache開源協議，同年改名為seata。

本文將入手seata，官方的文檔和demo主要以dubbo和springcloud體系的接入為主。本文選取springboot作為項目構建框架，快速構建示例。

 

環境說明

seata的發展還是比較快的，而版本的更新帶來的使用變化可能會導致文檔的過時。本文在閱讀官方提供的quickStart基礎上完成。為了過程順利最好保持環境版本一致，否則你可能得自己debug問題所在。

1. jdk1.8
2. mysql8.0.18
3. springboot 2.2.5.RELEASE
4. spring-cloud-alibaba-dependencies 2.2.0.RELEASE
5. seata-server v1.1.0

當然，版本並不一定需要完全一樣。比如你可以使用mysql5+，但是就得強制指定對應的mysql-connector-java.jar的版本。

 

步驟說明

要完成這個示例項目，需要不少的步驟。這里提前羅列一下，比較心里有數

1. 搭建springboot項目
2. 引入seata依賴
3. 配置
   1. 添加並修改file.conf和registry.conf配置
   2. 添加數據源配置
   3. 數據源添加undo_log表
4. 測試
   1. 數據源添加業務表和數據
   2. 編寫業務代碼
   3. 全局回滾測試

 

搭建springboot項目

搭建springboot項目比較簡單，本文采用idea構建了兩個項目

1）user-service

application.properties配置為：

server.port=8080
server.servlet.context-path=/user-service
spring.application.name=user-service

 

2) good-service

server.port=8081
server.servlet.context-path=/good-service
spring.application.name=good-service

 

繼承自

 

基礎依賴為

 

引入seata依賴

首先Import一下dependencies，注意：groupId和官方文檔寫的不一樣，版本是2.2.0.RELEASE

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>com.alibaba.cloud</groupId>
            <artifactId>spring-cloud-alibaba-dependencies</artifactId>
            <version>2.2.0.RELEASE</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

 

接着引入seata的依賴

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
</dependency>

 

配置

添加並修改file.conf和registry.conf

我們需要在resources目錄下，創建file.conf和registry.conf這兩個文件。

registry.conf文件不需要修改，直接拷貝即可

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "file"

  nacos {
    serverAddr = "localhost"
    namespace = "public"
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://localhost:8761/eureka"
    application = "default"
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = "0"
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
  }
  consul {
    cluster = "default"
    serverAddr = "127.0.0.1:8500"
  }
  etcd3 {
    cluster = "default"
    serverAddr = "http://localhost:2379"
  }
  sofa {
    serverAddr = "127.0.0.1:9603"
    application = "default"
    region = "DEFAULT_ZONE"
    datacenter = "DefaultDataCenter"
    cluster = "default"
    group = "SEATA_GROUP"
    addressWaitTime = "3000"
  }
  file {
    name = "file.conf"
  }
}

config {
  # file、nacos 、apollo、zk、consul、etcd3
  type = "file"

  nacos {
    serverAddr = "localhost"
    namespace = "public"
    cluster = "default"
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    app.id = "seata-server"
    apollo.meta = "http://192.168.1.204:8801"
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    session.timeout = 6000
    connect.timeout = 2000
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}

 

file.conf內容如下，但是要service節點下的一個配置。我們示例項目是user-service和good-service，分別在對應的項目中要做修改。

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
  }
  shutdown {
    # when destroy server, wait seconds
    wait = 3
  }
  serialization = "seata"
  compressor = "none"
}
service {
  #vgroup->rgroup
  vgroup_mapping.取spring.application.name的值-seata-service-group = "default"
  #only support single node
  default.grouplist = "127.0.0.1:8091"
  #degrade current not support
  enableDegrade = false
  #disable
  disable = false
  #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent
  max.commit.retry.timeout = "-1"
  max.rollback.retry.timeout = "-1"
}

client {
  async.commit.buffer.limit = 10000
  lock {
    retry.internal = 10
    retry.times = 30
  }
  report.retry.count = 5
}

## 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
  }

  ## database store
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "dbcp"
    ## mysql/oracle/h2/oceanbase etc.
    db-type = "mysql"
    url = "jdbc:mysql://127.0.0.1:3306/seata"
    user = "mysql"
    password = "mysql"
    min-conn = 1
    max-conn = 3
    global.table = "global_table"
    branch.table = "branch_table"
    lock-table = "lock_table"
    query-limit = 100
  }
}
lock {
  ## the lock store mode: local、remote
  mode = "remote"

  local {
    ## store locks in user's database
  }

  remote {
    ## store locks in the seata's server
  }
}
recovery {
  committing-retry-delay = 30
  asyn-committing-retry-delay = 30
  rollbacking-retry-delay = 30
  timeout-retry-delay = 30
}

transaction {
  undo.data.validation = true
  undo.log.serialization = "jackson"
}

## metrics settings
metrics {
  enabled = false
  registry-type = "compact"
  # multi exporters use comma divided
  exporter-list = "prometheus"
  exporter-prometheus-port = 9898
}

 

添加數據源配置

分布式事務的實現，數據源代理是很重要的方式。在2.2.0.RELEASE中，數據源代理自動實現了，不需要我們去配置一個代理類。但是我們還是需要配置一下數據源的。

首先在application.properties中添加配置

user-service的配置

spring.datasource.driverClassName=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/db_user?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&useSSL=false
spring.datasource.username=root
spring.datasource.password=

good-service的配置

spring.datasource.driverClassName=com.mysql.cj.jdbc.Driver
spring.datasource.url=jdbc:mysql://localhost:3306/db_good?useUnicode=true&characterEncoding=utf8&allowMultiQueries=true&useSSL=false
spring.datasource.username=root
spring.datasource.password=

注意：你可能選擇了與本文不同的MySQL版本，那么driverClassName或許並不是com.mysql.cj.jdbc.Driver而是早期的com.mysql.jdbc.Driver 。否則會報驅動類找不到的問題

 

然后我們添加一個配置類，這里以druid數據源為例。

@Configuration
public class DataSourceProxyConf {

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource dataSource() {
        return new DruidDataSource();
    }
}

 

這樣，數據源就配置好了。后面的測試中，我們將會使用JdbcTemplate進行數據源操作，以及resttemplate作為服務調用。所以這里也順便配置兩個Bean吧

@Configuration
public class DataSourceProxyConf {

    @Bean
    @ConfigurationProperties(prefix = "spring.datasource")
    public DataSource dataSource() {
        return new DruidDataSource();
    }

    @Bean
    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
        return new JdbcTemplate(dataSource);
    }

    @Bean
    public RestTemplate restTemplate() {
        return new RestTemplate();
    }
}

這里要注意，jdbcTemplate注入的dataSource不是純粹的DruidDataSource，而是DataSourceProxy。前面我們說過，seata在2.2.0版本進行了自動代理，不需要像2.1.0那種配置代理對象了。

 

添加undo_log表

配置的最后一個項目，就是在兩個數據源中添加undo_log表。我們將兩個db分別命名為db_user和db_good吧。

undo_log表用於保存回滾數據，直接將以下sql在db里面執行一下即可。

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

 

到這里，配置的部分就結束了，后續進入測試的部分

 

測試

數據源添加業務表和業務數據

在db_user和db_good分別添加表t_user和t_good

CREATE TABLE `t_good` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `amount` int(255) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;

 

CREATE TABLE `t_user` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `account` decimal(10,2) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8 COLLATE=utf8_bin;

 

並添加數據如下

t_user

 

t_good

 

 

編寫業務代碼

數據准備好了，我們簡單編寫一下controller

GoodController對t_good表的amount字段-1操作，再1/0發生算術異常

@RestController
@RequestMapping("good")
public class GoodController {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @GetMapping("amount/reduce")
    public String reduceAmount() {
        jdbcTemplate.update("update t_good set amount = amount - 1 where id = 1");

        int i = 1/0;

        return "success";
    }
}

 

UserController先對t_user表的account字段-1操作，然后調用GoodController。

這里我們注意到@GlobalTransactional這個注解，表示開啟分布式事務。

@RestController
@RequestMapping("user")
public class UserController {

    @Autowired
    private JdbcTemplate jdbcTemplate;

    @Autowired
    private RestTemplate restTemplate;

    @GetMapping("account/reduce")
    @GlobalTransactional(rollbackFor = Exception.class) public String reduceAccount() {
        jdbcTemplate.update("update t_user set account = account - 1 where id = 1");

        restTemplate.getForEntity("http://localhost:8081/good-service/good/amount/reduce", String.class);
        return "success";
    }
}

 

全局回滾測試

1、我們先啟動seata-server，直接執行.seata-server.sh(mac或linux) 或者 seata-server.bat(windows)。seata-server下載地址為：https://github.com/seata/seata/releases/download/v1.1.0/seata-server-1.1.0.zip

2、接着啟動user-service和good-service

3、當調用接口：http://localhost:8080/user-service/user/account/reduce的時候會爆出500內部錯誤。這時候檢查一下數據源或者seata-server的console你會發現數據沒有變化，console出現了兩個branchId對應的doRollback輸出。再看看undo_log表，自增ID從1變成了2.

 

總結

本文到此結束了，簡單搭建並測試了一下commit和rollback。雖然阿里已經盡量把使用變得很簡單了，但是明顯的是搭建一個示例項目還是經歷了不少步驟。可見分布式項目帶來的成本降低，但是復雜度上升的困難是很難逾越的。

后續的文章中將從源碼角度了解seata的實現，雖然很麻煩，但是...莫名地其樂無窮吧~