建立站点的作用,住房城乡住房和城乡建设部网站,最新的销售平台,灰大设计导航网分布式事务seata的AT模式介绍
seata是阿里开源的一款分布式事务解决方案#xff0c;致力于提供高性能和简单易用的分布式事务服务。Seata 将为用户提供了 AT、TCC、SAGA 和 XA 事务模式#xff0c;本文主要介绍AT模式的使用。
seata安装
下载seata服务#xff0c;官方地址…分布式事务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 sessionStoremaxBranchSessionSize 16384maxGlobalSessionSize 512fileWriteBufferCacheSize 16384sessionReloadReadSize 100flushDiskMode async}db {datasource druiddbType mysqldriverClassName com.mysql.cj.jdbc.Driverurl jdbc:mysql://192.168.234.1:3306/seata?useUnicodetruecharacterEncodingutf8useSSLfalseserverTimezoneUTC //修改数据库连接user seata //修改数据库账号password 123456 //修改数据库密码minConn 5maxConn 30globalTable global_tablebranchTable branch_tablelockTable lock_tablequeryLimit 100}
}service {vgroup_mapping.chuanzh_tx_group default //chuanzh_tx_group为自定义的事务组名称要和客户端配置保持一致default.grouplist 192.168.234.128:8091enableDegrade falsedisable falsemax.commit.retry.timeout -1max.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)
) ENGINEInnoDB AUTO_INCREMENT1 DEFAULT CHARSETutf8;配置conf/registry.conf文件
registry {type eureka 修改注册方式微服务调用使用的是Eurekanacos {serverAddr localhostnamespace cluster default}eureka {serviceUrl http://192.168.234.1:8081/eureka //修改Eureka地址application default weight 1}redis {serverAddr localhost:6379db 0}zk {cluster defaultserverAddr 127.0.0.1:2181session.timeout 6000connect.timeout 2000}consul {cluster defaultserverAddr 127.0.0.1:8500}etcd3 {cluster defaultserverAddr http://localhost:2379}sofa {serverAddr 127.0.0.1:9603application defaultregion DEFAULT_ZONEdatacenter DefaultDataCentercluster defaultgroup SEATA_GROUPaddressWaitTime 3000}file {name file.conf}
}以上修改了使用Eureka方式注册并配置了Eureka地址启动MySQL、Eureka服务后就可以启动seata服务了。
调用方配置store-server
maven配置使用seata-spring-boot-starter自动配置的方式不需要再添加file.conf和register.conf文件 dependencygroupIdcom.alibaba/groupIdartifactIddruid-spring-boot-starter/artifactIdversion${druid-spring-boot-starter.version}/version/dependencydependencygroupIdio.seata/groupIdartifactIdseata-spring-boot-starter/artifactIdversion1.2.0/version/dependencyapplication.properties配置
server.port9090
spring.application.namestore-servermybatis.type-aliases-packagecom.chuanzh.model
mybatis.mapper-locationsclasspath:mapper/*.xmlspring.datasource.urljdbc:mysql://localhost:3306/test?useUnicodetruecharacterEncodingutf-8
spring.datasource.usernameroot
spring.datasource.password123456
spring.datasource.driver-class-namecom.mysql.jdbc.Driverseata.tx-service-groupchuanzh_tx_group
seata.service.vgroup-mapping.chuanzh_tx_groupdefault
seata.service.grouplist.default192.168.234.128:8091logging.level.io.seataDEBUGeureka.client.serviceUrl.defaultZone http://localhost:8081/eureka/数据源配置因为seata是对数据库的datasource进行了接管和代理所以在每个参与分布式事务的数据源都要进行如下配置
Configuration
public class DataSourceConfiguration {BeanConfigurationProperties(prefix spring.datasource)public DataSource druidDataSource(){DruidDataSource druidDataSource new DruidDataSource();return druidDataSource;}PrimaryBean(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数据
