Spring Cloud整合Seata实现分布式事务

Spring Cloud整合Seata分布式事务框架

Seata:阿里巴巴开源的一款分布式解决方案,其前身是Fescar。官网:https://seata.io/zh-cn/index.html。

1. 添加依赖

1<dependency> 2 <groupId>com.alibaba.cloud</groupId> 3 <artifactId>spring-cloud-alibaba-seata</artifactId> 4 <version>2.1.1.RELEASE</version> 5 </dependency>

2. 添加seata配置和注册文件

以下配置文件在要下载的seata-server/conf文件夹里有,直接从文件里复制作相应修改即可。

  • 注册文件 registry.conf:

该配置用于指定 TC 的注册中心和配置文件,默认使用file文件注册,则该文件无需修改。如果是注册到eureka,则要修改registry.type="eureka",并修改registry.eureka.serviceUrl为你当前eureka注册中心的地址。其它注册方式类推。

1registry { 2 # file 、nacos 、eureka、redis、zk、consul、etcd3、sofa 3 type = "file" 4 5 nacos { 6 serverAddr = "localhost" 7 namespace = "" 8 cluster = "default" 9 } 10 eureka { 11 serviceUrl = "http://localhost:8761/eureka" 12 application = "default" 13 weight = "1" 14 } 15 redis { 16 serverAddr = "localhost:6379" 17 db = "0" 18 } 19 zk { 20 cluster = "default" 21 serverAddr = "127.0.0.1:2181" 22 session.timeout = 6000 23 connect.timeout = 2000 24 } 25 consul { 26 cluster = "default" 27 serverAddr = "127.0.0.1:8500" 28 } 29 etcd3 { 30 cluster = "default" 31 serverAddr = "http://localhost:2379" 32 } 33 sofa { 34 serverAddr = "127.0.0.1:9603" 35 application = "default" 36 region = "DEFAULT_ZONE" 37 datacenter = "DefaultDataCenter" 38 cluster = "default" 39 group = "SEATA_GROUP" 40 addressWaitTime = "3000" 41 } 42 file { 43 name = "file.conf" 44 } 45} 46 47config { 48 # file、nacos 、apollo、zk、consul、etcd3 49 type = "file" 50 51 nacos { 52 serverAddr = "localhost" 53 namespace = "" 54 } 55 consul { 56 serverAddr = "127.0.0.1:8500" 57 } 58 apollo { 59 app.id = "seata-server" 60 apollo.meta = "http://192.168.1.204:8801" 61 } 62 zk { 63 serverAddr = "127.0.0.1:2181" 64 session.timeout = 6000 65 connect.timeout = 2000 66 } 67 etcd3 { 68 serverAddr = "http://localhost:2379" 69 } 70 file { 71 name = "file.conf" 72 } 73}
  • file.conf:

该配置用于指定TC的相关属性;如果使用注册中心也可以将配置添加到配置中心。
这里要注意的是transport.service.vgroup_mapping配置的值,在 Spring Cloud 中默认值是${spring.application.name}-fescar-service-group,可以通过指定application.propertiesspring.cloud.alibaba.seata.tx-service-group这个属性覆盖,但是必须要和file.conftransport.service.vgroup_mapping的值一致,否则会提示 no available server to connect

1transport { 2 # tcp udt unix-domain-socket 3 type = "TCP" 4 #NIO NATIVE 5 server = "NIO" 6 #enable heartbeat 7 heartbeat = true 8 #thread factory for netty 9 thread-factory { 10 boss-thread-prefix = "NettyBoss" 11 worker-thread-prefix = "NettyServerNIOWorker" 12 server-executor-thread-prefix = "NettyServerBizHandler" 13 share-boss-worker = false 14 client-selector-thread-prefix = "NettyClientSelector" 15 client-selector-thread-size = 1 16 client-worker-thread-prefix = "NettyClientWorkerThread" 17 # netty boss thread size,will not be used for UDT 18 boss-thread-size = 1 19 #auto default pin or 8 20 worker-thread-size = 8 21 } 22 shutdown { 23 # when destroy server, wait seconds 24 wait = 3 25 } 26 serialization = "seata" 27 compressor = "none" 28} 29service { 30 #vgroup->rgroup 31 vgroup_mapping.my_test_tx_group = "default" 32 #only support single node 33 default.grouplist = "127.0.0.1:8091" 34 #degrade current not support 35 enableDegrade = false 36 #disable 37 disable = false 38 #unit ms,s,m,h,d represents milliseconds, seconds, minutes, hours, days, default permanent 39 max.commit.retry.timeout = "-1" 40 max.rollback.retry.timeout = "-1" 41} 42 43client { 44 async.commit.buffer.limit = 10000 45 lock { 46 retry.internal = 10 47 retry.times = 30 48 } 49 report.retry.count = 5 50 tm.commit.retry.count = 1 51 tm.rollback.retry.count = 1 52} 53 54## transaction log store 55store { 56 ## store mode: file、db 57 mode = "file" 58 59 ## file store 60 file { 61 dir = "sessionStore" 62 63 # branch session size , if exceeded first try compress lockkey, still exceeded throws exceptions 64 max-branch-session-size = 16384 65 # globe session size , if exceeded throws exceptions 66 max-global-session-size = 512 67 # file buffer size , if exceeded allocate new buffer 68 file-write-buffer-cache-size = 16384 69 # when recover batch read size 70 session.reload.read_size = 100 71 # async, sync 72 flush-disk-mode = async 73 } 74 75 ## database store 76 db { 77 ## the implement of javax.sql.DataSource, such as DruidDataSource(druid)/BasicDataSource(dbcp) etc. 78 datasource = "dbcp" 79 ## mysql/oracle/h2/oceanbase etc. 80 db-type = "mysql" 81 driver-class-name = "com.mysql.jdbc.Driver" 82 url = "jdbc:mysql://127.0.0.1:3306/seata" 83 user = "mysql" 84 password = "mysql" 85 min-conn = 1 86 max-conn = 3 87 global.table = "global_table" 88 branch.table = "branch_table" 89 lock-table = "lock_table" 90 query-limit = 100 91 } 92} 93lock { 94 ## the lock store mode: local、remote 95 mode = "remote" 96 97 local { 98 ## store locks in user's database 99 } 100 101 remote { 102 ## store locks in the seata's server 103 } 104} 105recovery { 106 #schedule committing retry period in milliseconds 107 committing-retry-period = 1000 108 #schedule asyn committing retry period in milliseconds 109 asyn-committing-retry-period = 1000 110 #schedule rollbacking retry period in milliseconds 111 rollbacking-retry-period = 1000 112 #schedule timeout retry period in milliseconds 113 timeout-retry-period = 1000 114} 115 116transaction { 117 undo.data.validation = true 118 undo.log.serialization = "jackson" 119 undo.log.save.days = 7 120 #schedule delete expired undo_log in milliseconds 121 undo.log.delete.period = 86400000 122 undo.log.table = "undo_log" 123} 124 125## metrics settings 126metrics { 127 enabled = false 128 registry-type = "compact" 129 # multi exporters use comma divided 130 exporter-list = "prometheus" 131 exporter-prometheus-port = 9898 132} 133 134support { 135 ## spring 136 spring { 137 # auto proxy the DataSource bean 138 datasource.autoproxy = false 139 } 140}

3. 配置数据源

Seata 通过代理数据源的方式实现分支事务;MyBatis 和 JPA 都需要注入 io.seata.rm.datasource.DataSourceProxy, 此外,MyBatis 还需要额外注入 org.apache.ibatis.session.SqlSessionFactory。

  • MyBatis

    @Configuration public class DataSourceProxyConfig {

    1@Bean 2@ConfigurationProperties(prefix = "spring.datasource") 3public DataSource dataSource() { 4 return new DruidDataSource(); 5} 6 7@Bean 8public DataSourceProxy dataSourceProxy(DataSource dataSource) { 9 return new DataSourceProxy(dataSource); 10} 11 12@Bean 13public SqlSessionFactory sqlSessionFactoryBean(DataSourceProxy dataSourceProxy) throws Exception { 14 SqlSessionFactoryBean sqlSessionFactoryBean = new SqlSessionFactoryBean(); 15 sqlSessionFactoryBean.setDataSource(dataSourceProxy); 16 sqlSessionFactoryBean.setTransactionFactory(new SpringManagedTransactionFactory()); 17 return sqlSessionFactoryBean.getObject(); 18}

    }

特别提示:对于3的2个配置文件和4的数据源配置,每个参与分布式事务的服务都需要引入,要在一个事务组里。

4. 添加Seata需要用到的undo_log

该sql在seata-server/conf里有提供。该表用于在分布式事务发生异常时执行回滚的依据。每个参与分布式事务的数据库都需要加该表

1-- the table to store seata xid data 2-- 0.7.0+ add context 3-- you must to init this sql for you business databese. the seata server not need it. 4-- 此脚本必须初始化在你当前的业务数据库中,用于AT 模式XID记录。与server端无关(注:业务数据库) 5-- 注意此处0.3.0+ 增加唯一索引 ux_undo_log 6drop table `undo_log`; 7CREATE TABLE `undo_log` ( 8 `id` bigint(20) NOT NULL AUTO_INCREMENT, 9 `branch_id` bigint(20) NOT NULL, 10 `xid` varchar(100) NOT NULL, 11 `context` varchar(128) NOT NULL, 12 `rollback_info` longblob NOT NULL, 13 `log_status` int(11) NOT NULL, 14 `log_created` datetime NOT NULL, 15 `log_modified` datetime NOT NULL, 16 `ext` varchar(100) DEFAULT NULL, 17 PRIMARY KEY (`id`), 18 UNIQUE KEY `ux_undo_log` (`xid`,`branch_id`) 19) ENGINE=InnoDB AUTO_INCREMENT=1 DEFAULT CHARSET=utf8;

5. 下载运行seata-server服务

下载地址:https://seata.io/zh-cn/blog/download.html,根据需要下载相应版本。解压后,seata/conf目录有以上需要的配置等。修改相应的配置,启动运行。双击seata/bin目录下的seata-server.bat或者执行以下命令运行:

sh ./bin/seata-server.sh

6. 添加注解@GlobalTransactional开启分布式事务

1@Override 2 @GlobalTransactional 3 public void create(String userId, String commodityCode, int orderCount) { 4 int orderMoney = calculate(commodityCode, orderCount); 5 Order order = new Order(); 6 order.setUserId(userId); 7 order.setCommodityCode(commodityCode); 8 order.setCount(orderCount); 9 order.setMoney(orderMoney); 10 orderMapper.insert(order); 11 12 // Feign远程调用,扣钱 13 accountFeignClient.debit(userId, orderMoney); 14 // 减库存 15 storageFeignClient.deduct(commodityCode, orderCount); 16 17 if (orderCount == 3) { 18 throw new RuntimeException("异常回滚"); 19 } 20 }

7. 测试

  • 正常测试

http://localhost:8081/order/create?userId=1&commodityCode=book&orderCount=1

  • 异常测试

http://localhost:8081/order/create?userId=1&commodityCode=book&orderCount=3

这里我代码里写了orderCount==3时候会抛异常,用于测试。另外,我们也可以去掉@GlobalTransactional测试。

8. 详细代码

点赞
收藏

评论区

加载中...

相关推荐

MySQL:[Err] 1292 - Incorrect datetime value: ‘0000-00-00 00:00:00‘ for column ‘CREATE_TIME‘ at row 1

文章目录问题用navicat导入数据时,报错:原因这是因为当前的MySQL不支持datetime为0的情况。解决修改sql\mode:sql\mode:SQLMode定义了MySQL应支持的SQL语法、数据校验等,这样可以更容易地在不同的环境中使用MySQL。全局s

Oracle 分组与拼接字符串同时使用

SELECTT.,ROWNUMIDFROM(SELECTT.EMPLID,T.NAME,T.BU,T.REALDEPART,T.FORMATDATE,SUM(T.S0)S0,MAX(UPDATETIME)CREATETIME,LISTAGG(TOCHAR(

MySQL部分从库上面因为大量的临时表tmp_table造成慢查询

背景描述Time:20190124T00:08:14.70572408:00User@Host:@Id:Schema:sentrymetaLast_errno:0Killed:0Query_time:0.315758Lock_

手写Java HashMap源码

HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程HashMap的使用教程22

2020年前端实用代码段,为你的工作保驾护航

有空的时候,自己总结了几个代码段,在开发中也经常使用,谢谢。1、使用解构获取json数据let jsonData  id: 1,status: "OK",data: 'a', 'b';let  id, status, data: number   jsonData;console.log(id, status, number )

Twitter的分布式自增ID算法snowflake (Java版)

概述分布式系统中,有一些需要使用全局唯一ID的场景,这种时候为了防止ID冲突可以使用36位的UUID,但是UUID有一些缺点,首先他相对比较长,另外UUID一般是无序的。有些时候我们希望能使用一种简单一些的ID,并且希望ID能够按照时间有序生成。而twitter的snowflake解决了这种需求,最初Twitter把存储系统从MySQL迁移