SpringCloud 整合分布式事務元件 Seata (八)

前言

近期一直在忙項目，我也是打工仔。不多說，我們開始玩一玩seata。

正文

什麼都不說，我們按照慣例，先上一個圖（圖裡不規範的使用請忽略）：

簡單一眼就看出來， 比我們平時用的東西，多了  Seata Server 微服務 。

同樣這個 Seata Server 微服務  ，也是需要注冊到eureka上面去的。 

那麼我們首先就搞一搞這個 seata server ，那麼剩下的就是一些原本的業務服務整合配置了。

該篇用的 seata server 版本，用的是1.4.1 ， 可以去git下載下傳下。

當然，我也是給你們備了的：

seata server 1.4.1

某度網盤分享位址：

https://pan.baidu.com/s/1R9McfkSkoj72Pf98ugCvBw

提取碼：

4ynp

第一步，下載下傳下來解壓 ：

第二步，創個 seata server 用的資料庫 ：

CREATE TABLE `branch_table` (
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(128) NOT NULL,
  `transaction_id` bigint(20) DEFAULT NULL,
  `resource_group_id` varchar(32) DEFAULT NULL,
  `resource_id` varchar(256) DEFAULT NULL,
  `branch_type` varchar(8) DEFAULT NULL,
  `status` tinyint(4) DEFAULT NULL,
  `client_id` varchar(64) DEFAULT NULL,
  `application_data` varchar(2000) DEFAULT NULL,
  `gmt_create` datetime DEFAULT NULL,
  `gmt_modified` datetime DEFAULT NULL,
  PRIMARY KEY (`branch_id`),
  KEY `idx_xid` (`xid`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
CREATE TABLE `global_table` (
  `xid` varchar(128) NOT NULL,
  `transaction_id` bigint(20) DEFAULT NULL,
  `status` tinyint(4) NOT NULL,
  `application_id` varchar(32) DEFAULT NULL,
  `transaction_service_group` varchar(32) DEFAULT NULL,
  `transaction_name` varchar(128) DEFAULT NULL,
  `timeout` int(11) DEFAULT NULL,
  `begin_time` bigint(20) DEFAULT NULL,
  `application_data` varchar(2000) DEFAULT NULL,
  `gmt_create` datetime DEFAULT NULL,
  `gmt_modified` datetime DEFAULT NULL,
  PRIMARY KEY (`xid`),
  KEY `idx_gmt_modified_status` (`gmt_modified`,`status`),
  KEY `idx_transaction_id` (`transaction_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
 
CREATE TABLE `lock_table` (
  `row_key` varchar(128) NOT NULL,
  `xid` varchar(96) DEFAULT NULL,
  `transaction_id` bigint(20) DEFAULT NULL,
  `branch_id` bigint(20) NOT NULL,
  `resource_id` varchar(256) DEFAULT NULL,
  `table_name` varchar(32) DEFAULT NULL,
  `pk` varchar(36) DEFAULT NULL,
  `gmt_create` datetime DEFAULT NULL,
  `gmt_modified` datetime DEFAULT NULL,
  PRIMARY KEY (`row_key`),
  KEY `idx_branch_id` (`branch_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
           

建立完後：

第三步，修改下 \seata-1.4.1\seata-server-1.4.1\seata\conf 裡的配置檔案一些資訊 ：

1. registry.conf

ok，registry.conf 這檔案就修改這些配置項。

2. file.conf ：

以上兩個檔案配置完（記得儲存）， 我們先把我們的注冊中心 eureka服務跑起來，然後點選啟動 seata server：

可以看到啟動成功（前提是eureka已經啟動）：

第三步 ，配置我們需要用到 分布式事務 seata元件的 微服務 ：

我這裡的示例實踐，需要用到的有2個微服務 ：

那麼我們這兩個微服務都需要做點什麼呢？

1. 在對應的微服務的對應的不同資料庫裡(隻要你想用上seata的)， 都加上undo_log 這個表：

SQL語句：

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;

-- ----------------------------
-- Table structure for undo_log
-- ----------------------------
DROP TABLE IF EXISTS `undo_log`;
CREATE TABLE `undo_log`  (
  `id` bigint(20) NOT NULL AUTO_INCREMENT,
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `context` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `rollback_info` longblob NOT NULL,
  `log_status` int(11) NOT NULL,
  `log_created` datetime(0) NULL,
  `log_modified` datetime(0) NULL,
  PRIMARY KEY (`id`) USING BTREE,
  UNIQUE INDEX `ux_undo_log`(`xid`, `branch_id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Dynamic;

SET FOREIGN_KEY_CHECKS = 1;
           

2. 關鍵步驟了 ， 導入jar包 （需要用到seata元件都服務都需要導入）

<dependency>
			<groupId>com.alibaba.cloud</groupId>
			<artifactId>spring-cloud-alibaba-seata</artifactId>
			<version>2.1.0.RELEASE</version>
			<exclusions>
				<exclusion>
					<artifactId>seata-all</artifactId>
					<groupId>io.seata</groupId>
				</exclusion>
			</exclusions>
		</dependency>
		<dependency>
			<artifactId>seata-all</artifactId>
			<groupId>io.seata</groupId>
			<version>1.4.1</version>
		</dependency>
           

3.更加關鍵了，就是做配置

先總得了解下，需要用到seata的 微服務需要做些什麼配置 ？

1. 在resources 下 新增2個配置檔案 ， file.conf  和   registry.conf

2.yml 新增配置seata 事務組參數

3.代碼調整資料源代理，交給seata代理

1.   registry.conf  

以上是業務微服務裡的registry.conf  需要改動的配置資訊 ，給出一份該篇文章使用的：

registry {
  # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa
  type = "eureka"
  loadBalance = "RandomLoadBalance"
  loadBalanceVirtualNodes = 10

  nacos {
    application = "seata-server"
    serverAddr = "127.0.0.1:8848"
    group = "SEATA_GROUP"
    namespace = ""
    cluster = "default"
    username = ""
    password = ""
  }
  eureka {
    serviceUrl = "http://localhost:8761/eureka/"
    application = "seata-server"
    weight = "1"
  }
  redis {
    serverAddr = "localhost:6379"
    db = 0
    password = ""
    cluster = "default"
    timeout = 0
  }
  zk {
    cluster = "default"
    serverAddr = "127.0.0.1:2181"
    sessionTimeout = 6000
    connectTimeout = 2000
    username = ""
    password = ""
  }
  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 = "127.0.0.1:8848"
    namespace = ""
    group = "SEATA_GROUP"
    username = ""
    password = ""
  }
  consul {
    serverAddr = "127.0.0.1:8500"
  }
  apollo {
    appId = "seata-server"
    apolloMeta = "http://192.168.1.204:8801"
    namespace = "application"
    apolloAccesskeySecret = ""
  }
  zk {
    serverAddr = "127.0.0.1:2181"
    sessionTimeout = 6000
    connectTimeout = 2000
    username = ""
    password = ""
  }
  etcd3 {
    serverAddr = "http://localhost:2379"
  }
  file {
    name = "file.conf"
  }
}
           

file.conf

給出一份該篇使用的完整的：

transport {
  # tcp udt unix-domain-socket
  type = "TCP"
  #NIO NATIVE
  server = "NIO"
  #enable heartbeat
  heartbeat = true
  # the client batch send request enable
  enableClientBatchSendRequest = true
  #thread factory for netty
  threadFactory {
    bossThreadPrefix = "NettyBoss"
    workerThreadPrefix = "NettyServerNIOWorker"
    serverExecutorThread-prefix = "NettyServerBizHandler"
    shareBossWorker = false
    clientSelectorThreadPrefix = "NettyClientSelector"
    clientSelectorThreadSize = 1
    clientWorkerThreadPrefix = "NettyClientWorkerThread"
    # netty boss thread size,will not be used for UDT
    bossThreadSize = 1
    #auto default pin or 8
    workerThreadSize = "default"
  }
  shutdown {
    # when destroy server, wait seconds
    wait = 3
  }
  serialization = "seata"
  compressor = "none"
}
service {
#這裡注意,等号前後都是配置,前面是yml裡配置的事務組,後面是register.conf裡定義的seata-server
  vgroupMapping.test_tx_group = "seata-server"
  #only support when registry.type=file, please don't set multiple addresses
  seata_tc_server.grouplist = "127.0.0.1:8091"
  #degrade, current not support
  enableDegrade = false
  #disable seata
  disableGlobalTransaction = false
}
client {
  rm {
    asyncCommitBufferLimit = 10000
    lock {
      retryInterval = 10
      retryTimes = 30
      retryPolicyBranchRollbackOnConflict = true
    }
    reportRetryCount = 5
    tableMetaCheckEnable = false
    reportSuccessEnable = false
  }
  tm {
    commitRetryCount = 5
    rollbackRetryCount = 5
  }
  undo {
    dataValidation = true
    logSerialization = "jackson"
    logTable = "undo_log"
  }
  log {
    exceptionRate = 100
  }
}
           

2. 需要在yml配置檔案加上配置項，指明目前服務使用了 seata分布式事務元件，且需要加入的分布式事務組是哪個： 

spring:
  cloud:
    alibaba:
      seata.tx-service-group: test_tx_group
           

3.然後是需要将資料源交給seata去代理：

去掉預設自動加載資料源

配置dao層掃描位置

@MapperScan("com.cloud.client1.dao")
@SpringBootApplication(exclude = DataSourceAutoConfiguration.class)
           

然後是seata代理資料源（紅色标注的地方需要注意下，都是實打實踩出來的坑。導入類的來源以及掃描的mapper的xml位置）：

ok，到這裡一個微服務 usercent的 整合 seata 分布式元件算是完成了。 

同樣，我們繼續對 另外一個微服務 coredata 做seata 分布式元件  整合 。

也是第一步，file.conf 和 registry.conf  （其實我們現在就是讓這兩個微服務在同一個分布式事務組，我們直接把剛才在usercent那邊的兩個檔案直接粘貼過來就好）。

然後第二步就是修改yml配置：

第三步就是新增seata資料源代理、application上的注解排除自動加載資料源和掃描dao層位址，yml上面的配置項：

完事，兩個業務微服務都整合了 分布式事務元件 seata ，而且都設定在同個分布式事務組裡 （test_tx_group）.

那麼到這其實已經完事了， 接下來玩下 分布式事務復原場景 的例子 (怎麼去使用)。

PS： 使用方式關注 @GlobalTransactional  注解， 想把全局事務從哪裡開啟，就把這個注解放到哪個方法上去， 這個方法内 調用的其他服務隻要都使用了seata，且在同個事務組，那麼就會加入到目前的全局事務裡。 

我該篇就不弄太多服務了，就弄了2個分布式服務。   

展示的例子内容 ：

1. 上遊服務 出錯， 觸發分布式事務， 上下遊服務都會事務復原；

2.下遊服務 出錯，觸發分布式事務， 上下遊服務都會事務復原；（某種程度上講，下遊出錯如果隻有一個下遊，其實不需分布式事務，通過錯誤傳遞單個復原也是可以的。但是如果服務調用鍊很長，中遊服務出錯，需要整個鍊上的服務都事務復原，那麼就有必要都使用seata ）

我們開始模拟： 

第一個場景

上遊服務 coredata 使用seata全局事務注解@GlobalTransactional 标記方法，先插入一個Account資料 ；

然後調用下遊服務 usercent 插入一條資料；

然後 上遊服務 coredata繼續執行業務邏輯，繼續插入資料；

接着模拟上遊服務coredata開始報錯（我們通過name長度故意觸發錯誤）；

期望結果：  上下遊兩個服務 在目前方法事務下插入的資料都復原！

上遊服務 coredata 方法：

通過fegin調用下遊服務 usercent 方法：

下遊服務 usercent 的插入方法：

開始模拟:

1.先把eureka跑起來：

暫時就注冊中心自己，沒有别的服務:

2.把seata server 服務跑起來，注冊到eureka上去：

我是window環境，執行.bat

可以看到已經成功注冊到eureka上了：

3. 把上遊微服務 coredata 和 下遊服務 usercent 都跑起來：

然後看下seata server上，也可以看到 兩個服務 都成功‘注冊’到了 seata server上了，而且都在同一個事務組 test_tx_group裡面：

接下來，就是開始調用一下我們模拟的場景代碼就完事了（不過我會加點圖來給大家簡單分析下）：

一開始，都是沒有資料的：

調用上遊服務coredata接口觸發一下整個流程：

調用開始:

我們打斷點到 上遊已經插入過一次資料，下遊也插入過一次資料，

可以看到上遊服務和下遊服務的資料庫裡面的undo_log表出現了 事務記錄，有關目前事務的 branch_id和 所在事務 xid，而且可以看到在兩個服務内的undo_log表中記錄的xid都是一緻的，代表他們都在一個事務中：

然後我們繼續往下執行，故意讓上遊報錯：

這時候，接口調用完畢結束：

我們看到seata server裡面的資訊，可以看到全局事務 xid為22080結尾，復原成功：

復原成功後，undo_log表中的記錄會删除掉：

當然我們兩個服務裡面也是沒有資料的，因為復原了：

這裡可能有人會想，你查一下是空就能證明是復原了麼？ 

這時候我們也可以利用主鍵自增目前值可以看到确實發生了資料復原的場景：

第一個場景就到此吧。

接下來我們模拟第二個場景：

上遊服務 coredata 使用seata全局事務注解@GlobalTransactional 标記方法，先插入一個Account資料 ；

然後調用下遊服務 usercent 插入一條資料；

然後下遊服務 直接模拟出錯， 這樣觸發事務復原。

期望結果：  上下遊兩個服務 在目前方法事務下插入的資料都復原！

也就是說我們需要對下遊服務的插入方法裡面做手腳，故意抛出錯誤：

快速調用一下：

可以看到下遊出錯了：

看下我們的seata server怎麼說,已出發分布式事務，復原成功：

資料庫裡面的資料也是復原了，空的： 

好了，該篇springboot cloud使用eureka整合  分布式事務元件  Seata  就到此吧。

ps： 最近比較忙，每篇文章其實都是用一些零碎時間拼湊出來的。不過我會堅持我的文章的初衷，能讓大家跟着實踐，能搞懂，能學會。

我...隻是個散工。