领先百科

  1. 首页 > 知识库 > >

分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

生活百科

前言在上一篇博文（分布式事务与Seate框架（1）——分布式事务理论）中了解了足够的分布式事务的理论知识后，到了实践部分，在工作中虽然用到了Seata，但是自己却并没有完全实践过，所以自己私下花点时间实践以加深理解，实际上在实践过程中遇到了很多的坑（比如Seata与SpringCloudAlibaba的整合中版本兼容性问题，是个很让人头疼的一块，甚至专门去github提过issue），有时候甚至得跟踪源码进行分析，这也使我加强了对阅读源码的能力 。总之都是要code的 。本篇博文主要结合实践深入讲解Seata AT模式！
参考资料《Spring Cloud Alibaba 微服务原理与实战》（PDF电子书资源，有需要的小伙伴可以评论私信我）、官方wiki

博文中源码已上传至github(https://github.com/Jian0110/learning-cloudalibaba)，欢迎小伙伴们star...
一、实践准备工作1、框架介绍实践主要是以“订单-库存-账户”系统演示，主要的框架图如下，图中各个部分充当的分布式事务角色已标明 。

分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
具体流程：
1）用户登录XXX商品购物系统（假设已有账户），
2）点击购买某个商品，发起创建订单请求；
3）检查购买商品的库存量，如果不够则创建订单失败提示库存不足；否则锁定该商品---->减少库存--->创建订单；
4）订单创建成功后点击付款（或直接付款无需点击，实际上整个Demo中下单之后模拟立马支付，并不会点击付款）；
5）如果购买成功则对账户进行余额进行判断，余额足够则进行减扣，余额不够则进行提示说明
6）返回购买成功失败提示说明 。
2、项目结构项目结构如下：
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
mvn package打包运行seata服务，即运行TC服务器（这里只展示单机）
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
初始化Seata库，导入sql脚本
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
二、代码实践这里只展示关键代码，全部代码已提交gituhb：，有需要的小伙伴可以自行获取
1、“订单-库存-账户”服务订单服务：
TM（microService）：seata-order-service
RM（DB Resources）：jdbc:mysql://127.0.0.1:3306/order
OrderService：
@GlobalTransactional // TM开启全局事务@Transactional(rollbackFor = Exception.class)public void createOrder(Long productId, BigDecimal price){// 这里模拟获取的是用户的账户ID// 通过上下文获取userId再获取accountId（单个账户）Long accountId = 1L; // 假设已经获取到了账户ID// 1.rpc调用库存微服务检查库存并减库存操作Boolean deductStorageSuccess =storageFeignClient.deduct(productId);if (!deductStorageSuccess) {throw new RuntimeException("storage deduct failed!");}// 2.创建订单ProductOrder order =ProductOrder.builder().productId(productId).accountId(accountId).payAmount(price).build();log.info("create order : {}", order);// 这里为了模拟回滚，所以先对价格的判断放到了创建订单之后，抛出runtime exceptionif (price.compareTo(BigDecimal.ZERO) < 0) {throw new NumberFormatException("product price must greater than zero!");}orderMapper.insertSelective(order);// 3.rpc调用账户微服务对余额检查并扣款操作Boolean deductAccountSuccess =accountFeignClient.deduct(accountId, price);if (!deductAccountSuccess) {throw new RuntimeException("account deduct failed!");}// 4. 反馈结果}OrderController：
/*** 模拟创建订单* @param productId* @param price* @return*/@PostMapping("/create")public String create(Long productId, BigDecimal price){try {orderService.createOrder(productId, price);} catch (Exception e) {log.error("order failed: ", e);return "order failed";}return "order success";}调用的Feign：
@FeignClient(name="seata-account-service")public interface AccountFeignClient {@PostMapping("/account/deduct")Boolean deduct(@RequestParam("accountId") Long accountId, @RequestParam("payAmount") BigDecimal payAmount);}@FeignClient(name="seata-storage-service")public interface StorageFeignClient {@PostMapping("/storage/deduct")Boolean deduct(@RequestParam("productId") Long productId);}库存服务：
microService：seata-storage-service
RM（DB Resources）：jdbc:mysql://127.0.0.1:3306/storage
StorageService
public Boolean deduct(Long productId){// 这里先检查有没有库存了, 生产环境下这里是需要for update数据库锁，或者分布式锁Repo repoFromDB =repoMapper.selectByPrimaryKey(productId);if (repoFromDB == null) {throw new RuntimeException("product not exist!");}// 对库存减一int afterCount = repoFromDB.getAmount()-1;// 没有库存剩余了if (afterCount < 0) {throw new RuntimeException("product storage is no remaining!");}Repo repo = Repo.builder().id(productId).amount(afterCount).build();repoMapper.updateAmount(repo);log.info("deduct product[{}] storage, current amount is {}", productId, afterCount);return true;}StorageController:
/*** 模拟对商品库存减一* @param productId* @return*/@PostMapping("/deduct")public Boolean deduct(Long productId){try {storageService.deduct(productId);} catch (Exception e) {return false;}return true;}账户服务：
microService：seata-account-service
RM（DB Resources）：jdbc:mysql:127.0.0.1/account
AccountService：
public void deduct(Long accountId, BigDecimal payAmount){// 这里先检查有没有账户存在, 生产环境下这里是需要for update数据库锁，或者分布式锁UserAccount userAccountFromDB =userAccountMapper.selectByPrimaryKey(accountId);if (userAccountFromDB == null) {throw new RuntimeException("account not exist!");}// 检查余额是否足够BigDecimal afterBalance = userAccountFromDB.getBalance().subtract(payAmount);if (afterBalance.compareTo(BigDecimal.ZERO) < 0) {throw new RuntimeException("the balance is not enough!");}UserAccount userAccount = UserAccount.builder().id(accountId).balance(afterBalance).build();log.info("deduct account[{}] , current balance is {}", accountId, afterBalance);userAccountMapper.updateBalance(userAccount);}AccountController：
/*** 模拟账户扣款* @param accountId* @param payAmount* @return*/@PostMapping("/deduct")public Boolean deduct(Long accountId, BigDecimal payAmount){try {accountService.deduct(accountId, payAmount);} catch (Exception e) {return false;}return true;}2、Seata服务器，即TC角色首先初始化seata的sql脚本（sql脚本参考官方wiki），并开启seata库,之后开启Seata Server（具体的配置与启动前nacos配置，事务分组等相关概念请参考官方wiki）
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
3、检查Nacos服务与配置列表微服务模块启动后快速注册到dev命名空间下的SEATA_GROUP分组，此时TM、RM、TC都已经具备　
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
　  
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
启动微服务模块后可以看到日志输出，说明启动成功并且已经成功注册
RM will register :jdbc:mysql://127.0.0.1:3306/account
nacos registry, SEATA_GROUP seata-account-service 192.168.99.1:6009 register finishedStarted SeataAccountApplication in 30.115 seconds (JVM running for 33.158).......NettyPool create channel to transactionRole:TMROLE,address:169.254.6.29:8091,msg:< RegisterTMRequest{applicationId='seata-account-service', transactionServiceGroup='my_test_tx_group'} >register TM success. client version:1.4.0, server version:1.4.0,channel:[id: 0xa77dc065, L:/169.254.6.29:52794 - R:/169.254.6.29:8091]register success, cost 4 ms, version:1.4.0,role:TMROLE,channel:[id: 0xa77dc065, L:/169.254.6.29:52794 - R:/169.254.6.29:8091]三、运行测试1、模拟购买支付成功情况运行启动所有的微服务后，在TC Serve的日志可以看到所有的TM、RM都已经注册了
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
此时productId=1库存还剩998
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
accountId=1的用户余额还剩1000元
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
 接下来就是模拟用户购买商品环节，调用http://localhost:6008/order/create，表示用户想买商品ID=1，价格为12.25的商品
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
清空日志，并发起请求查看日志：
16:10:45.167INFO --- [rverHandlerThread_1_4_500] i.s.s.coordinator.DefaultCoordinator: Begin new global transaction applicationId: seata-order-service,transactionServiceGroup: my_test_tx_group, transactionName: createOrder(java.lang.Long, java.math.BigDecimal),timeout:60000,xid:169.254.6.29:8091:13613974712390860816:10:45.964INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: SeataMergeMessage xid=169.254.6.29:8091:136139747123908608,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/storage,lockKey=repo:1,clientIp:169.254.6.29,vgroup:my_test_tx_group16:10:46.086INFO --- [rverHandlerThread_1_5_500] i.seata.server.coordinator.AbstractCore: Register branch successfully, xid = 169.254.6.29:8091:136139747123908608, branchId = 136139750928142336, resourceId = jdbc:mysql://127.0.0.1:3306/storage ,lockKeys = repo:116:10:46.788INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: SeataMergeMessage xid=169.254.6.29:8091:136139747123908608,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/account,lockKey=user_account:1,clientIp:169.254.6.29,vgroup:my_test_tx_group16:10:46.918INFO --- [rverHandlerThread_1_6_500] i.seata.server.coordinator.AbstractCore: Register branch successfully, xid = 169.254.6.29:8091:136139747123908608, branchId = 136139754342305793, resourceId = jdbc:mysql://127.0.0.1:3306/account ,lockKeys = user_account:116:10:47.015INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: xid=169.254.6.29:8091:136139747123908608,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/order,lockKey=product_order:6,clientIp:169.254.6.29,vgroup:my_test_tx_group16:10:47.073INFO --- [rverHandlerThread_1_7_500] i.seata.server.coordinator.AbstractCore: Register branch successfully, xid = 169.254.6.29:8091:136139747123908608, branchId = 136139755294412801, resourceId = jdbc:mysql://127.0.0.1:3306/order ,lockKeys = product_order:616:10:47.184INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: xid=169.254.6.29:8091:136139747123908608,extraData=https://tazarkount.com/read/null,clientIp:169.254.6.29,vgroup:my_test_tx_group16:10:48.084INFO --- [AsyncCommitting_1_1] io.seata.server.coordinator.DefaultCore: Committing global transaction is successfully done, xid = 169.254.6.29:8091:136139747123908608.16:10:53.908INFO --- [TxTimeoutCheck_1_1] i.s.s.coordinator.DefaultCoordinator: Global transaction[169.254.6.29:8091:136139530647490560] is timeout and will be rollback.16:10:54.947INFO --- [RetryRollbacking_1_1] io.seata.server.coordinator.DefaultCore: Rollback global transaction successfully, xid = 169.254.6.29:8091:136139530647490560.从日志中我们可以看到：
1）全局事务XID已经生成，各个分支注册成功，
2）branchId也已经生成并在全局事务XID下，资源已被锁住
3）全局事务提交成功
查看此时的库存与余额，都已经进行了减扣
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图

分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
2、模拟库存不足情况修改productId=1的商品库存为0：
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
再次发起请求，查看TC Server日志，可以查出明显发生了全局事务的回滚
16:20:24.258INFO --- [verHandlerThread_1_12_500] i.s.s.coordinator.DefaultCoordinator: Begin new global transaction applicationId: seata-order-service,transactionServiceGroup: my_test_tx_group, transactionName: createOrder(java.lang.Long, java.math.BigDecimal),timeout:60000,xid:169.254.6.29:8091:13614217625087590416:20:24.279INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: xid=169.254.6.29:8091:136142176250875904,extraData=https://tazarkount.com/read/null,clientIp:169.254.6.29,vgroup:my_test_tx_group16:20:24.420INFO --- [verHandlerThread_1_13_500] io.seata.server.coordinator.DefaultCore: Rollback global transaction successfully, xid = 169.254.6.29:8091:136142176250875904.查看库存与余额情况，库存仍然是0，余额仍然是987.75
3、模拟余额不足情况修改accountId=1的账户余额小于12.25
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图
再次发起请求，查看日志
16:27:41.811INFO --- [verHandlerThread_1_14_500] i.s.s.coordinator.DefaultCoordinator: Begin new global transaction applicationId: seata-order-service,transactionServiceGroup: my_test_tx_group, transactionName: createOrder(java.lang.Long, java.math.BigDecimal),timeout:60000,xid:169.254.6.29:8091:13614401145600819216:27:41.836INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: SeataMergeMessage xid=169.254.6.29:8091:136144011456008192,branchType=AT,resourceId=jdbc:mysql://127.0.0.1:3306/storage,lockKey=repo:1,clientIp:169.254.6.29,vgroup:my_test_tx_group16:27:41.889INFO --- [verHandlerThread_1_15_500] i.seata.server.coordinator.AbstractCore: Register branch successfully, xid = 169.254.6.29:8091:136144011456008192, branchId = 136144011762192385, resourceId = jdbc:mysql://127.0.0.1:3306/storage ,lockKeys = repo:116:27:42.088INFO --- [batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler: xid=169.254.6.29:8091:136144011456008192,extraData=https://tazarkount.com/read/null,clientIp:169.254.6.29,vgroup:my_test_tx_group16:27:42.632INFO --- [verHandlerThread_1_16_500] io.seata.server.coordinator.DefaultCore: Rollback branch transaction successfully, xid = 169.254.6.29:8091:136144011456008192 branchId = 13614401176219238516:27:42.754INFO --- [verHandlerThread_1_16_500] io.seata.server.coordinator.DefaultCore: Rollback global transaction successfully, xid = 169.254.6.29:8091:136144011456008192.不同于库存不足的情况的是，这里库存服务分支事务是先注册TC Server的，因为有异常的并不是库存服务，需要注意的是因为我模拟的是下单之后立马支付，支付失败的话订单也是不会存在，实际生活中应该是订单显示“支付失败” 。
查看库存与余额情况，库存仍然是997，余额仍然是10.75
【分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践】
分布式事务与分布式锁 2 分布式事务与Seate框架——Seata实践

文章插图


    上一篇：java四种锁机制 Java四种引用类型原理你真的搞明白了吗？五分钟带你深入理解！

    下一篇：同事写了一个疯狂的类构造器，我要疯了，Builder 模式都不会么？！