单节点新老系统之分布式事务处理上篇-五八三

我正在参加「掘金·启航计划」

大家好！我是锦视，从事IT技术开发近12年，参与及独立负责过大大小小的项目不少于60个，接触过个各行各业，具有丰富的项目开发及管理经验，来掘金平台目的是记录自己职业生涯中的点点滴滴~~~

背景

由于公司老系统使用的技术比较陈旧，不能适应当今市场的需求，必须重构老系统来满足各方面的需求扩展，除了重构需要技术选型、开发规范、梳理需求等外，迫切需要解决的是新老系统数据库事务的问题，故此引入了此篇文章。

概念

3个What

什么是事务？

事务是并发控制的单位，是用户定义的一个操作序列。这是比较官方的回答，白话就是说把你的操作统一化，要么所有操作都成功，要么就都不成功，如果执行中有某一项操作失败，其之前所有的操作都回滚到未执行这一系列操作之前的状态。其具有以下四个特性：

原子性(Atomicity)：事务是数据库的逻辑工作单位，事务中包括的操作要么全做，要么全不做；
一致性(Consistency)：事务执行的结果必须是使数据库从一个一致性状态变到另一个一致性状态。一致性与原子性是密切相关的；
隔离性(Isolation)：一个事务的执行不能被其他事务干扰；
持续性/永久性(Durability)：一个事务一旦提交，它对数据库中数据的改变就应该是永久性的。

以上就是对事务的一个简单介绍，脏读、幻读、不可重复读等内容不在本篇范围内，请自行查看相关资料。

什么是分布式？

分布式是指在相同空间或不同空间的相同物体或不同物体之间的分布情况，这句话大家可能看不懂或不够通俗易懂，举个生活中的例子，比如：房屋，同一个客厅（空间）会有不同物件且分布在不同的位置，而客厅与房间又是不同的空间，客厅可能放有椅子房间也可能有（相同物件），客厅有沙发房间可以没有（不同物件）等，物件相等于是各自独立的模块，客厅和房间是服务器（节点），模块和服务器之间必然是有联系的，由此，引入了分布式系统的概念，分布式系统是指由单个或多个节点不同模块或相同模块组件的系统，既然是系统，比如模块间是有联系的。其中，节点指的是计算机服务器，而且这些节点一般不是孤立的，而是互通的，这些连通的节点上部署了我们的系统模块，并且相互的操作会有协同。

以上就是对分布式的简单理解，其他更详细内容不在本篇范围内，请自行查看相关资料。

什么是分布式事务？

分布式事务是指异构系统之间保证操作数据库事务的原子性、一致性、隔离性及永久性。

以上就是对分布式事务简单的介绍，其他更详细内容不在本篇范围内，请自行查看相关资料。

原理

众所周知，单个或多个节点中的系统要保证事务的ACID特性，必须要有一个事务协调中心，即全局事务控制，单个节点的本地事务在提交时必须上报给事务协调中心，由事务协调中心统一控制，中间会涉及到加锁和解锁的过程，保证并发问题，以下为原理图：

短事务

长事务

以上为单节点多系统原理图，画的不够详细，该图会不断的更新，如有不妥之处请指正，谢谢！

实战

前提
创建global_tx（事务控制表）、stock（库存表）、account（账户表），以下为建表语句：

DROP TABLE IF EXISTS `global_tx`;
CREATE TABLE `global_tx`  (
  `id` bigint(0) NOT NULL AUTO_INCREMENT,
  `type` varchar(50) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `tx_id` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `content` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `table` varchar(100) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `remark` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `po_type` varchar(50) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `po_id` bigint(0) NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;


DROP TABLE IF EXISTS `stock`;
CREATE TABLE `stock`  (
  `id` bigint(0) NOT NULL AUTO_INCREMENT,
  `product_id` bigint(0) NULL DEFAULT NULL,
  `amount` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `remark` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;


DROP TABLE IF EXISTS `account`;
CREATE TABLE `account`  (
  `id` bigint(0) NOT NULL AUTO_INCREMENT,
  `name` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  `cost` decimal(10, 2) NULL DEFAULT NULL,
  `remark` varchar(255) CHARACTER SET utf8 COLLATE utf8_bin NULL DEFAULT NULL,
  PRIMARY KEY (`id`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 2 CHARACTER SET = utf8 COLLATE = utf8_bin ROW_FORMAT = Dynamic;

其中global_tx为核心表，其他两个为测试表

短事务
- GlobalTxMapper

注：获取锁，根据要更新的数据id和表名去事务控制表中查找，如果能找到，则说明该数据已被锁定，不能修改

注：锁定被更新数据，不然其他事务拿到锁

注：释放锁，更新成功后释放锁，让其他事务可以拿到锁

StockMapper

注：更新库存（测试）

AccountMapper

注：银行转账（测试）

StockServiceImpl（核心代码）

// 重试次数
private static volatile int retryCount = 0;
......
@Transactional(rollbackFor = Exception.class, isolation = Isolation.READ_COMMITTED)
@Override
public void updateById(Long poId) throws Exception{
    Integer recordCount = txMapper.selectByCondition(poId, "stock");
     // 1. 判断是否能获取到锁
    if (Objects.isNull(recordCount) || 0 == recordCount) {
        // 1.1 拿到锁
        // 锁定
        GlobalTx gt = new GlobalTx();
        gt.setPoId(poId);
        gt.setContent("amount：1->10");
        gt.setPoType("update");
        gt.setType("SHORT_TRANSACTION");
        gt.setTxId(UUID.randomUUID().toString());
        gt.setTable("stock");
        gt.setRemark("更新库存");
        txMapper.insert(gt);
        // 更新库存
         mapper.updateById(poId);
         // 释放锁
        txMapper.delete(poId, "stock");
    }else {
        // 2. 未拿到锁
        // 模拟定时任务，不够严谨，睡眠1秒
        retryCount ++;
        // 最大重试3次
        if (retryCount > 3) {
            // TODO: 记录回滚日志
            throw new Exception("id为【"+poId+"】的库存已被锁定，不能修改！！！");
        }
        Thread.sleep(1);
        updateById(poId);
    }
}

注：启动类或事务配置类中需开启事务，注解为@EnableTransactionManagement（类注解）

如果锁定了某行记录，修改会报以下错误：

企业微信截图_16874181659403.png