springboot cloud使用eureka整合分佈式事務組件Seata 的方法

Posted on by

前言

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

正文

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

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

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

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

該篇用的 seata server 版本，用的是1.4.1 ， 可以去git下載下。當然，我也是給你們備瞭的：

seata server 1.4.1 某度網盤分享地址：

鏈接: https://pan.baidu.com/s/1Oj1NkKwU4jeLjJ3Pu9hT2Q

提取碼: 9at6

第一步，下載下來解壓 ：

第二步，創個 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）.

那麼到這其實已經完事瞭， 接下來玩下 分佈式事務回滾場景 的例子 (怎麼去使用)。

我該篇就不弄太多服務瞭，就弄瞭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： 最近比較忙，每篇文章其實都是用一些零碎時間拼湊出來的。不過我會堅持我的文章的初衷，能讓大傢跟著實踐，能搞懂，能學會。我…隻是個散工。

到此這篇關於SpringCloud 整合分佈式事務組件 Seata的文章就介紹到這瞭,更多相關SpringCloud 整合分佈式內容請搜索WalkonNet以前的文章或繼續瀏覽下面的相關文章希望大傢以後多多支持WalkonNet！

推薦閱讀：