当前位置： 正文

Seata框架的分布式事务实践

2021-05-16 技术推文 6.13k 推广

摘要

实践是理论的升华，我深刻体会到了这一点。虽然工作中用到了Seata，但我还是花时间自己实践了一遍，感觉收获颇丰。实践过程中，有欣喜、有挑战，更有成就感。

正文

分布式事务与Seate框架（2）——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、框架介绍

实践主要是以“订单-库存-账户”系统演示，主要的框架图如下，图中各个部分充当的分布式事务角色已标明。

　　　　

　　具体流程：

　　1）用户登录XXX商品购物系统（假设已有账户），

　　2）点击购买某个商品，发起创建订单请求；

　　3）检查购买商品的库存量，如果不够则创建订单失败提示库存不足；否则锁定该商品—->减少库存—>创建订单；

　　4）订单创建成功后点击付款（或直接付款无需点击，实际上整个Demo中下单之后模拟立马支付，并不会点击付款）；

　　5）如果购买成功则对账户进行余额进行判断，余额足够则进行减扣，余额不够则进行提示说明

　　6）返回购买成功失败提示说明。

2、项目结构

项目结构如下：

mvn package打包运行seata服务，即运行TC服务器（这里只展示单机）

初始化Seata库，导入sql脚本

二、代码实践

这里只展示关键代码，全部代码已提交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 exception
        if (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）

　　　　

3、检查Nacos服务与配置列表

　　微服务模块启动后快速注册到dev命名空间下的SEATA_GROUP分组，此时TM、RM、TC都已经具备　　  　　　　

启动微服务模块后可以看到日志输出，说明启动成功并且已经成功注册

RM will register :jdbc:mysql://127.0.0.1:3306/account
nacos registry, SEATA_GROUP seata-account-service 192.168.99.1:6009 register finished
Started 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都已经注册了

此时productId=1库存还剩998

accountId=1的用户余额还剩1000元

 接下来就是模拟用户购买商品环节，调用http://localhost:6008/order/create，表示用户想买商品ID=1，价格为12.25的商品

清空日志，并发起请求查看日志：

16:10:45.167  INFO --- [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:136139747123908608
16:10:45.964  INFO --- [     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_group
16:10:46.086  INFO --- [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:1
16:10:46.788  INFO --- [     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_group
16:10:46.918  INFO --- [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:1
16:10:47.015  INFO --- [     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_group
16:10:47.073  INFO --- [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:6
16:10:47.184  INFO --- [     batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler         : xid=169.254.6.29:8091:136139747123908608,extraData=null,clientIp:169.254.6.29,vgroup:my_test_tx_group
16:10:48.084  INFO --- [      AsyncCommitting_1_1] io.seata.server.coordinator.DefaultCore  : Committing global transaction is successfully done, xid = 169.254.6.29:8091:136139747123908608.
16:10:53.908  INFO --- [       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.947  INFO --- [     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、模拟库存不足情况

修改productId=1的商品库存为0：

再次发起请求，查看TC Server日志，可以查出明显发生了全局事务的回滚

16:20:24.258  INFO --- [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:136142176250875904
16:20:24.279  INFO --- [     batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler         : xid=169.254.6.29:8091:136142176250875904,extraData=null,clientIp:169.254.6.29,vgroup:my_test_tx_group
16:20:24.420  INFO --- [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

再次发起请求，查看日志

16:27:41.811  INFO --- [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:136144011456008192
16:27:41.836  INFO --- [     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_group
16:27:41.889  INFO --- [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:1
16:27:42.088  INFO --- [     batchLoggerPrint_1_1] i.s.c.r.p.server.BatchLogHandler         : xid=169.254.6.29:8091:136144011456008192,extraData=null,clientIp:169.254.6.29,vgroup:my_test_tx_group
16:27:42.632  INFO --- [verHandlerThread_1_16_500] io.seata.server.coordinator.DefaultCore  : Rollback branch transaction successfully, xid = 169.254.6.29:8091:136144011456008192 branchId = 136144011762192385
16:27:42.754  INFO --- [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

关注不迷路

扫码下方二维码，关注宇凡盒子公众号，免费获取最新技术内幕！

温馨提示：如果您访问和下载本站资源，表示您已同意只将下载文件用于研究、学习而非其他用途。
文章版权声明 1、本网站名称：宇凡盒子
2、本站文章未经许可，禁止转载！
3、如果文章内容介绍中无特别注明，本网站压缩包解压需要密码统一是：yufanbox.com
4、本站仅供资源信息交流学习，不保证资源的可用及完整性，不提供安装使用及技术服务。点此了解
5、如果您发现本站分享的资源侵犯了您的权益，请及时通知我们，我们会在接到通知后及时处理！提交入口
赏 0

大家在看

物联网流量卡在IOS手机上无法分享个人热点问题解决办法
技术推文
WIFI热点个人热点开启WIFI热点

苹果手机无法开启个人热点是因为其APN接入点默认为空。要解决这个问题，需要在设置中添加物联卡的APN。下面是设置步骤：确认物联卡已激活并开通数据流量，进入设置-移动蜂窝网络-蜂窝移动数据网络-移动数据网络，点击右上角+号添加APN，按照提供的APN信息填写并保存即可。这样就能顺利地开启个人热点了。
2024-04-27 641
微信使用指南：保护账号安全，学会玩转微信！
技术推文
微信养号

众所周知，微信的重要性不言而喻，如果不幸被封，对我们的工作将带来巨大的影响。因此，今天我给大家带来一套相关的教程，从选择手机和手机卡开始，到注册微信号，再到养号和处理被封的方式，都会详细介绍。这套教程非常全面且细致，正如磨刀不误砍柴工一样，学起来吧！
2023-12-03 697
最新申请Openai，GPT4.0 API Key完整详细图文教程
技术推文
ChatGPTChatGPT官网CVV安全码

最新动态： OpenAI在官网宣布，GPT-4 API全面开放使用。 GPT宣布只要有过付费...
2023-11-15 1.12k
测试5个配音软件，我选了这个免费软件。
技术推文
微软工具tts腾讯智影配音神器

这几天我研究了5款配音软件，最终找到了让自己满意的免费软件。现在我来分享一下，供网友参考，一起交流链接。
2023-11-14 677
小红书未实名账号手机号释放方法
技术推文
小红书小红书换绑手机号释放

以前在平台上搜索不到自己账号的内容，很可能是因为系统限制了流量。现在你需要考虑的是，你的账号可能会被封禁。
2023-11-02 923
京东 Plus 会员限时活动，99 元即可回本，双 11 送券秒杀。
技术推文
京东PLUS活动线报

京东活动已于24号凌晨开始，现在只需99元即可享受Plus会员特权，还可获得50-40洗衣券和50元超市卡。此外，双11即将到来，屯个会员绝对划算！
2023-10-25 613

评论0

请先

站点公告

🚀 【宇凡盒子】全网资源库转储中心

👉 注册即送VIP权限👈

👻 全站资源免费下载✅，欢迎注册！

记得 【收藏】+【关注】 谢谢！~~~

立即注册
没有账号？注册  忘记密码？

社交账号快速登录