分佈式事務seata的AT模式介紹

seata是阿里開源的一款分佈式事務解決方案，致力於提供高性能和簡單易用的分佈式事務服務。Seata 將為用戶提供了 AT、TCC、SAGA 和 XA 事務模式，本文主要介紹AT模式的使用。

seata安裝

下載seata服務，官方地址：https://github.com/seata/seata/releases
在Linux下，下載完成後，直接解壓，通過命令安裝即可：

sh ./bin/seata-server.sh

支持的啟動參數

參數	全寫	作用	備註
-h	--host	指定在註冊中心註冊的 IP	不指定時獲取當前的 IP，外部訪問部署在雲環境和容器中的 server 建議指定
-p	--port	指定 server 啟動的端口	默認為 8091
-m	--storeMode	事務日誌存儲方式	支持file和db，默認為 file
-n	--serverNode	用於指定seata-server節點ID	,如 1,2,3..., 默認為 1
-e	--seataEnv	指定 seata-server 運行環境	如 dev, test 等, 服務啟動時會使用 registry-dev.conf 這樣的配置

如：

sh ./bin/seata-server.sh -p 8091 -h 127.0.0.1 -m file

seata的AT模式介紹

AT模式實質是兩階段提交協議的演變，具體如下：

• 一階段：業務數據和回滾日誌記錄在同一個本地事務中提交，釋放本地鎖和連接資源
• 二階段：
  提交異步化，非常快速地完成。

回滾通過一階段的回滾日誌進行反向補償。

業務背景：
用戶調用系統A的store服務，store服務調用系統B的company服務，company服務會新增一條數據，然後把companyId返回系統A，然後系統A通過companyId再新增一條store數據。

一般如果store服務執行失敗了，直接拋異常了，所以company服務也不會執行，
但如果store服務執行成功了，已經寫了一條數據到數據庫，執行company服務時失敗了，就會產生數據不一致的問題。

使用seata的AT模式，主要分為下面幾個步驟：

• 配置seata服務及創建事務表
• 調用方配置（對應上面的store服務）
• 服務提供方配置（對應上面的company服務）

配置seata服務及創建事務表

配置conf/file.conf文件

## transaction log store, only used in server side
store {
  ## store mode: file、db
  mode = "db" //修改為db模式，標識事務信息用db存儲
  ## file store property
  file {
    ## store location dir
    dir = "sessionStore"
    # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions
    maxBranchSessionSize = 16384
    # globe session size , if exceeded throws exceptions
    maxGlobalSessionSize = 512
    # file buffer size , if exceeded allocate new buffer
    fileWriteBufferCacheSize = 16384
    # when recover batch read size
    sessionReloadReadSize = 100
    # async, sync
    flushDiskMode = async
  }

  ## database store property
  db {
    ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc.
    datasource = "druid"
    ## mysql/oracle/postgresql/h2/oceanbase etc.
    dbType = "mysql"
    driverClassName = "com.mysql.cj.jdbc.Driver"
    url = "jdbc:mysql://192.168.234.1:3306/seata?useUnicode=true&characterEncoding=utf8&useSSL=false&&serverTimezone=UTC" //修改數據庫連接
    user = "seata" //修改數據庫賬號
    password = "123456" //修改數據庫密碼
    minConn = 5
    maxConn = 30
    globalTable = "global_table"
    branchTable = "branch_table"
    lockTable = "lock_table"
    queryLimit = 100
  }
}
## server configuration, only used in server side
service {
  #vgroup->rgroup
  vgroup_mapping.chuanzh_tx_group = "default" //chuanzh_tx_group為自定義的事務組名稱，要和客戶端配置保持一致
  #only support single node
  default.grouplist = "192.168.234.128: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"
}

上面配置共修改了3個地方：

1. 存儲模式改為db模式，需要創建3張事務表，如下：

   -- the table to store GlobalSession data
    CREATE TABLE IF NOT EXISTS `global_table`
    (
        `xid`                       VARCHAR(128) NOT NULL,
        `transaction_id`            BIGINT,
        `status`                    TINYINT      NOT NULL,
        `application_id`            VARCHAR(32),
        `transaction_service_group` VARCHAR(32),
        `transaction_name`          VARCHAR(128),
        `timeout`                   INT,
        `begin_time`                BIGINT,
        `application_data`          VARCHAR(2000),
        `gmt_create`                DATETIME,
        `gmt_modified`              DATETIME,
        PRIMARY KEY (`xid`),
        KEY `idx_gmt_modified_status` (`gmt_modified`, `status`),
        KEY `idx_transaction_id` (`transaction_id`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8;
   
    -- the table to store BranchSession data
    CREATE TABLE IF NOT EXISTS `branch_table`
    (
        `branch_id`         BIGINT       NOT NULL,
        `xid`               VARCHAR(128) NOT NULL,
        `transaction_id`    BIGINT,
        `resource_group_id` VARCHAR(32),
        `resource_id`       VARCHAR(256),
        `branch_type`       VARCHAR(8),
        `status`            TINYINT,
        `client_id`         VARCHAR(64),
        `application_data`  VARCHAR(2000),
        `gmt_create`        DATETIME(6),
        `gmt_modified`      DATETIME(6),
        PRIMARY KEY (`branch_id`),
        KEY `idx_xid` (`xid`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8;
   
    -- the table to store lock data
    CREATE TABLE IF NOT EXISTS `lock_table`
    (
        `row_key`        VARCHAR(128) NOT NULL,
        `xid`            VARCHAR(96),
        `transaction_id` BIGINT,
        `branch_id`      BIGINT       NOT NULL,
        `resource_id`    VARCHAR(256),
        `table_name`     VARCHAR(32),
        `pk`             VARCHAR(36),
        `gmt_create`     DATETIME,
        `gmt_modified`   DATETIME,
        PRIMARY KEY (`row_key`),
        KEY `idx_branch_id` (`branch_id`)
    ) ENGINE = InnoDB
      DEFAULT CHARSET = utf8;
   

2. 修改數據庫連接，注意如果你安裝的是MySQL8，則需要修改MySQL8的驅動：driverClassName = "com.mysql.cj.jdbc.Driver"，不然會出現啟動報錯的問題，詳細請參考：seata啟動MySQL報錯 #359。
3. 修改事務的組名，你也可以不修改，我這裡使用的是：chuanzh_tx_group

4. 創建業務事務表，記錄業務需要回滾的數據，在分佈式事務中，每個參與的業務數據庫都需要添加對應的表

   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;

配置conf/registry.conf文件

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "eureka"  修改註冊方式，微服務調用使用的是Eureka

  nacos {
    serverAddr = "localhost"
    namespace = ""
    cluster = "default"
  }
  eureka {
    serviceUrl = "http://192.168.234.1:8081/eureka"  //修改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"
  }
}

以上修改了使用Eureka方式註冊，並配置了Eureka地址，啟動MySQL、Eureka服務後，就可以啟動seata服務了。

調用方配置（store-server）

maven配置，使用seata-spring-boot-starter，自動配置的方式，不需要再添加file.conf和register.conf文件

    <!--druid-->
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>druid-spring-boot-starter</artifactId>
        <version>${druid-spring-boot-starter.version}</version>
    </dependency>

    <!--seata-->
    <dependency>
        <groupId>io.seata</groupId>
        <artifactId>seata-spring-boot-starter</artifactId>
        <version>1.2.0</version>
    </dependency>

application.properties配置：

server.port=9090
spring.application.name=store-server

mybatis.type-aliases-package=com.chuanzh.model
mybatis.mapper-locations=classpath:mapper/*.xml

spring.datasource.url=jdbc:mysql://localhost:3306/test?useUnicode=true&characterEncoding=utf-8
spring.datasource.username=root
spring.datasource.password=123456
spring.datasource.driver-class-name=com.mysql.jdbc.Driver

#注意這裡的事務組配置要和服務端一致
seata.tx-service-group=chuanzh_tx_group
seata.service.vgroup-mapping.chuanzh_tx_group=default
seata.service.grouplist.default=192.168.234.128:8091

logging.level.io.seata=DEBUG
## eureka
eureka.client.serviceUrl.defaultZone= http://localhost:8081/eureka/

數據源配置，因為seata是對數據庫的datasource進行了接管和代理，所以在每個參與分佈式事務的數據源都要進行如下配置：

@Configuration
public class DataSourceConfiguration {

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

    @Primary
    @Bean("dataSource")
    public DataSourceProxy dataSource(DataSource druidDataSource){
        return new DataSourceProxy(druidDataSource);
    }

    @Bean
    public SqlSessionFactory sqlSessionFactory(DataSourceProxy dataSourceProxy)throws Exception{
        SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean();
        sqlSessionFactoryBean.setDataSource(dataSourceProxy);
        sqlSessionFactoryBean.setMapperLocations(new PathMatchingResourcePatternResolver()
                .getResources("classpath*:/mapper/*.xml"));
        sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory());
        return sqlSessionFactoryBean.getObject();
    }

}

注意配置了數據源後，還需要在啟動類排除dataSource自動配置，不然會出現循環依賴的問題,如下，其它的解決方法，可以參考：集成fescar數據源循環依賴錯誤解決方案

@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)

配置請求攔截器，生成一個請求事務ID，用於在微服務中傳遞

@Configuration
public class SeataRequestInterceptor implements RequestInterceptor {
    @Override
    public void apply(RequestTemplate requestTemplate) {
        String xid = RootContext.getXID();
        if (StringUtils.isNotBlank(xid)) {
            //構建請求頭
            requestTemplate.header("TX_XID", xid);
        }
    }
}

服務提供方配置（company-server）

maven、application.properties、數據源配置同調用方配置，區別主要是攔截器的配置，如下：

@Slf4j
@Component
public class SeataHandlerInterceptor implements HandlerInterceptor {

    public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
        String xid = RootContext.getXID();
        String rpcXid = request.getHeader("TX_XID");
        //獲取全局事務編號
        if(log.isDebugEnabled()) {
            log.debug("xid in RootContext {} xid in RpcContext {}", xid, rpcXid);
        }
        if(xid == null && rpcXid != null) {
            //設置全局事務編號
            RootContext.bind(rpcXid);
            if(log.isDebugEnabled()) {
                log.debug("bind {} to RootContext", rpcXid);
            }
        }
        return true;
    }
    public void afterCompletion(HttpServletRequest request, HttpServletResponse response, Object handler, Exception e) {
        String rpcXid = request.getHeader("TX_XID");
        if(!StringUtils.isEmpty(rpcXid)) {
            String unbindXid = RootContext.unbind();
            if(log.isDebugEnabled()) {
                log.debug("unbind {} from RootContext", unbindXid);
            }

            if(!rpcXid.equalsIgnoreCase(unbindXid)) {
                log.warn("xid in change during RPC from {} to {}", rpcXid, unbindXid);
                if(unbindXid != null) {
                    RootContext.bind(unbindXid);
                    log.warn("bind {} back to RootContext", unbindXid);
                }
            }

        }
    }

}

@Configuration
public class WebMvcConfiguration implements WebMvcConfigurer {

    @Autowired
    private SeataHandlerInterceptor seataHandlerInterceptor;

    public void addInterceptors(InterceptorRegistry registry) {
        //註冊HandlerInterceptor，攔截所有請求
        registry.addInterceptor(seataHandlerInterceptor).addPathPatterns(new String[]{"/**"});
    }

}

添加全局事務註解

在服務調用方的方法上添加@GlobalTransactional註解，下面模擬了一種場景，如果companyId為偶數，則會拋異常。

    @GlobalTransactional(rollbackFor = Exception.class)
    public void create(StoreEntity storeEntity) throws Exception {
        CompanyEntity companyEntity = new CompanyEntity();
        companyEntity.setName(storeEntity.getName());
        companyEntity = companyFeign.createCompany(companyEntity);

        /**
         * 模擬異常
         */
        if (companyEntity.getId() % 2 == 0) {
            throw new Exception();
        }

        /** 寫入store數據 */
        storeEntity.setCompanyId(companyEntity.getId());
        storeMapper.insert(storeEntity);
    }

經過測試，companyFeign.createCompany服務調用後會先向數據庫寫一條數據，當create方法執行拋異常，就會事務回滾，刪除掉原先的company數據