分布式事務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文件
store {mode = "db" //修改為db模式,標識事務信息用db存儲file {dir = "sessionStore"maxBranchSessionSize = 16384maxGlobalSessionSize = 512fileWriteBufferCacheSize = 16384sessionReloadReadSize = 100flushDiskMode = async}db {datasource = "druid"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 = 5maxConn = 30globalTable = "global_table"branchTable = "branch_table"lockTable = "lock_table"queryLimit = 100}
}service {vgroup_mapping.chuanzh_tx_group = "default" //chuanzh_tx_group為自定義的事務組名稱,要和客戶端配置保持一致default.grouplist = "192.168.234.128:8091"enableDegrade = falsedisable = falsemax.commit.retry.timeout = "-1"max.rollback.retry.timeout = "-1"
}
上面配置共修改了3個地方:
-
存儲模式改為db模式,需要創建3張事務表,如下:
-- the table to store GlobalSession dataCREATE 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 = InnoDBDEFAULT CHARSET = utf8;-- the table to store BranchSession dataCREATE 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 = InnoDBDEFAULT CHARSET = utf8;-- the table to store lock dataCREATE 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 = InnoDBDEFAULT CHARSET = utf8;
-
修改數據庫連接,注意如果你安裝的是MySQL8,則需要修改MySQL8的驅動:driverClassName = “com.mysql.cj.jdbc.Driver”,不然會出現啟動報錯的問題,詳細請參考:seata啟動MySQL報錯 #359。
-
修改事務的組名,你也可以不修改,我這里使用的是:chuanzh_tx_group
-
創建業務事務表,記錄業務需要回滾的數據,在分布式事務中,每個參與的業務數據庫都需要添加對應的表
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 {type = "eureka" 修改注冊方式,微服務調用使用的是Eurekanacos {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 = 6000connect.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文件
<dependency><groupId>com.alibaba</groupId><artifactId>druid-spring-boot-starter</artifactId><version>${druid-spring-boot-starter.version}</version></dependency><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-servermybatis.type-aliases-package=com.chuanzh.model
mybatis.mapper-locations=classpath:mapper/*.xmlspring.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.Driverseata.tx-service-group=chuanzh_tx_group
seata.service.vgroup-mapping.chuanzh_tx_group=default
seata.service.grouplist.default=192.168.234.128:8091logging.level.io.seata=DEBUGeureka.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);}@Beanpublic 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 {@Overridepublic 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 {@Autowiredprivate SeataHandlerInterceptor seataHandlerInterceptor;public void addInterceptors(InterceptorRegistry registry) {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();}storeEntity.setCompanyId(companyEntity.getId());storeMapper.insert(storeEntity);}
經過測試,companyFeign.createCompany服務調用后會先向數據庫寫一條數據,當create方法執行拋異常,就會事務回滾,刪除掉原先的company數據