分布式事务(Seata)

阅读数:114 评论数:0

跳转到新版页面

分类

python/Java

正文

一、概述

1、单体应用

当系统的体量很小时，单体架构完全可以满足现有业务需求，所有的业务共有一个数据库，这时并不需要考虑分布式事务。

2、微服务

可随着业务量的不断增长，单体架构渐渐力不从心，此时就需要对数据库、表做分库分表处理，将应用服务化拆分（微服务）。可能会产生如订单中心、用户中心、库存中心等，由此带来的问题就是业务间相互隔离，每个业务都维护着自己的数据库，数据的交换只能通过RPC调用。

例如，当用户下单时，需同时对订单库、库存库、用户库进行操作，可此时我们只能保证自己本地的数据一致性，无法保证调用其他服务的操作是否成功，所以为了保证整个下单流程的一致性，就需要分布式事务介入。

3、CAP理论

（1）Consistency（一致性）：用户访问分布式系统中的任意节点，得到的数据必须一致。

（2）Availability（可用性）：访问任意健康节点，必须能得到响应，而不是超时或拒绝。

（3）Partition tolerance（分区容错）: 因为网络故障等原因导致分布式系统中的部分节点与其它节点失去通信，形成独立分区时，整个系统也要持续对外提供服务。

一个分布式系统不可能同时满足C A P三者。

AP模式：各子事务分辊执行和提交，允许出现结果不一致，然后采用措施恢复数据即可，实现最终一致。

CP模式：各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。

4、BASE理论

（1）Basically Available （基本可用）：分布式系统在出现故障时，允许损失部分可用性。

（2）Soft State（软状态）：在一定时间内，允许出现中间状态，比哪临时的不一致状态。

（3）Eventually Consisitent（最终一致性）：虽然无法保证一致性，但是在软状态结束后，最终达到数据一致。

二、分布式事务的方案

实现分布式事务的方案比较多，常见的比如基于XA协议的2PC、3PC，基于业务层的TCC，还有应用消息队列+消息表实现的最终一致性方案，还有一些中间件如Seata。

1、XA协议

XA协议是一种分布式事务规范，XA规范主要定义了全局的事务管理器（TM）和局部资源管理器（RM），本地数据库如mysql在XA中扮演的RM角色。

2、2PC

两阶段提交，对业务侵入很小，它最大的优势就是对使用方透明，用户可以使用本地事务一样使用基于XA协议的分布式事务。

（1）第一阶段（prepare）：即所有的参与者RM准备执行事务并锁住需要的资源，参与者ready时，向TM报告已准备就绪。

（2）第二阶段（commit/rollback）：当事务管理者（TM）确认所有参与者（RM）都ready后，向所有参与者发送commit命令。

目前主流的数据库基本都支持XA事务，包括mysql、oracle、sqlserver、postgre

我们看看本地数据库是如何支持XA的：

XA start '4fPqCNTYeSG'
UPDATE `user_account` SET `balance`=balance + 30,`update_time`='2021-06-09 11:50:42.438' WHERE user_id = '1'
XA end '4fPqCNTYeSG'
XA prepare '4fPqCNTYeSG'

当所有的参与者完成了prepare，就进入第二阶段 提交

xa commit '4fPqCNTYeSG'

2PC的缺点也是显而易见，它是一个强一致性的同步阻塞协议，事务执行过程中需将所有资源全部锁定，所以它比较适用于执行时间确定的短事务，整体性能比较差。

一旦事务协调者宕机或者发生网络抖动，会让参与者一直处于锁定资源的状态或者只有一部分参与提交成功，导致数据的不一致。因此，在高并发性能的场景中，2PC并不是一个很好的选择。

3、3PC

2PC中只有协调者有超时机制，3PC在协调者和参与者中都引入了超时机制，协调者出现故障后，参与者就不会一直阻塞。而且在第一阶段和第二阶段中又插入了一个准备阶段，保证了在最后提交之前各参与节点的状态是一致的。

虽然3PC用超时机制解决了协调者故障后参与者的阻塞问题，但与此同时却多了一次网络通信，性能上反而变得更差，了也太推荐。

4、TCC

TCC是两阶段提交的一种变种，它在业务层编写代码实现两阶段提交，它对业务的侵入性很强，难被利用。

5、消息事务（最终一致性）

消息事务其实就是基于消息中间件的两阶段提交，将本地事务和发消息放在同一个事务里，保证本地操作和发送消息同时成功。

基于消息中间件的两阶段提交方案，通常用在高并发场景下使用，牺牲数据的强一致性换取性能的大幅提升，不过实现这种方式的成本和复杂度是比较高的，还要看实际业务情况。

6、Seata

Seata也是从两阶段提交演变而来的一种分布式事务解决方案，提供了AT、TCC、SAGA和XA等事务模式 。

模式 XA AT TCC SAGA
一致性 强一致 弱一致 弱一致 最终一致
隔离性 完全隔离 基于全局锁隔离 基于资源预留隔离 无隔离
代码侵入 无 无 有，需要编写三个接口方法 有，需要编写状态机和补偿业务
性能 差 好 非常好 非常好
场景 对一致性、隔离性有高要求的业务 基于关系型数据库的大多数分布式事务场景都可以 对性能要求较高的事务。有非关系型数据库要参与的事务。 业务流程长、业务流程多。参与者包含其它公司或遗留系统服务，无法提供 TCC 模式要求的三个接口。

三、Seata（AT模式）

1、Seata分布式事务的几种角色

Transaction Coordinator(TC) 全局事务协调者，用来协调全局事务和各个分支事务（不同服务）的状态， 驱动全局事务和各个分支事务的回滚或提交。
Transaction Manager  事务管理者，业务层中用来开启/提交/回滚一个整体事务（在调用服务的方法中用注解开启事务）。
Resource Manager(RM)  资源管理者，一般指业务数据库代表了一个分支事务。管理分支事务与 TC 进行协调注册分支事务并且汇报分支事务的状态，驱动分支事务的提交或回滚。

2、第一阶段

在本地事务提交前，各分支事务需向全局事务协调者TC注册分支Branch Id，为要修改的记录申请全局锁，而如果一直拿不到锁那就需要回滚本地事务，TM开启事务后会生成全局唯一的XID，会在各个调用的服务间进行传递。

有了这样的机制，本地事务分支便可以在全局事务的第一阶段提交，并马上释放本地事务锁定的资源。相比于传统的XA，Seata降低了锁范围提高效率，即使第二阶段发生异常需要回滚，也可以快速从undo_log表中找到对应回滚数据并反解析成SQL来达到回滚补偿。

最后本地事务提交，业务数据的更新和前面生成的 UNDO LOG 数据一并提交，并将本地事务提交的结果上报给全局事务协调者 TC。

3、第二阶段

根据各分去的决议做提交或回滚。

如果决议是全局提交，此时各分支事务已提交成功，这时全局事务协调者TC会向分支发送第二阶段请求，这些请求会被放入一个异步任务队列中，异步队列会异步和批量地根据Branch ID查找并删除相应的undo log回滚记录。

如果决议是全局回滚，过程比全局提交麻烦一点，RM收到TC发来的回滚请求，通过XID和Branch ID找到相应的回滚日志记录，通过回滚记录生成反向更SQL并执行，以完成分支的回滚。

四、Seata使用

1、部署TC（Server端）

官方地址：  https://seata.io/zh-cn/blog/download.html

（1）修改配置registry.conf

这里将seata服务注册到nacos注册中心。

registry {
  # tc服务的注册中心类，这里选择nacos，也可以是eureka、zookeeper等
  type = "nacos"

  nacos {
    # seata tc 服务注册到 nacos的服务名称，可以自定义
    application = "seata-tc-server"
    serverAddr = "127.0.0.1:8848"
    group = "DEFAULT_GROUP"
    namespace = ""
    cluster = "HZ"
    username = "nacos"
    password = "nacos"
  }
}

config {
  # 读取tc服务端的配置文件的方式，这里是从nacos配置中心读取，这样如果tc是集群，可以共享配置
  type = "nacos"
  # 配置nacos地址等信息
  nacos {
    serverAddr = "127.0.0.1:8848"
    namespace = ""
    group = "SEATA_GROUP"
    username = "nacos"
    password = "nacos"
    dataId = "seataServer.properties"
  }
}

（2）在nacos上添加seataServer.properties

# 数据存储方式，db代表数据库
store.mode=db
store.db.datasource=druid
store.db.dbType=mysql
store.db.driverClassName=com.mysql.jdbc.Driver
store.db.url=jdbc:mysql://127.0.0.1:3306/seata?useUnicode=true&rewriteBatchedStatements=true
store.db.user=root
store.db.password=123
store.db.minConn=5
store.db.maxConn=30
store.db.globalTable=global_table
store.db.branchTable=branch_table
store.db.queryLimit=100
store.db.lockTable=lock_table
store.db.maxWait=5000

# 事务、日志等配置
server.recovery.committingRetryPeriod=1000
server.recovery.asynCommittingRetryPeriod=1000
server.recovery.rollbackingRetryPeriod=1000
server.recovery.timeoutRetryPeriod=1000
server.maxCommitRetryTimeout=-1
server.maxRollbackRetryTimeout=-1
server.rollbackRetryTimeoutUnlockEnable=false
server.undo.logSaveDays=7
server.undo.logDeletePeriod=86400000

# 客户端与服务端传输方式
transport.serialization=seata
transport.compressor=none
# 关闭metrics功能，提高性能
metrics.enabled=false
metrics.registryType=compact
metrics.exporterList=prometheus
metrics.exporterPrometheusPort=9898

（3）创建数据库表

SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;

-- ----------------------------
-- 分支事务表
-- ----------------------------
DROP TABLE IF EXISTS `branch_table`;
CREATE TABLE `branch_table`  (
  `branch_id` bigint(20) NOT NULL,
  `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `transaction_id` bigint(20) NULL DEFAULT NULL,
  `resource_group_id` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `resource_id` varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `branch_type` varchar(8) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `status` tinyint(4) NULL DEFAULT NULL,
  `client_id` varchar(64) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `application_data` varchar(2000) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `gmt_create` datetime(6) NULL DEFAULT NULL,
  `gmt_modified` datetime(6) NULL DEFAULT NULL,
  PRIMARY KEY (`branch_id`) USING BTREE,
  INDEX `idx_xid`(`xid`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;

-- ----------------------------
-- 全局事务表
-- ----------------------------
DROP TABLE IF EXISTS `global_table`;
CREATE TABLE `global_table`  (
  `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `transaction_id` bigint(20) NULL DEFAULT NULL,
  `status` tinyint(4) NOT NULL,
  `application_id` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `transaction_service_group` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `transaction_name` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `timeout` int(11) NULL DEFAULT NULL,
  `begin_time` bigint(20) NULL DEFAULT NULL,
  `application_data` varchar(2000) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `gmt_create` datetime NULL DEFAULT NULL,
  `gmt_modified` datetime NULL DEFAULT NULL,
  PRIMARY KEY (`xid`) USING BTREE,
  INDEX `idx_gmt_modified_status`(`gmt_modified`, `status`) USING BTREE,
  INDEX `idx_transaction_id`(`transaction_id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;

SET FOREIGN_KEY_CHECKS = 1;

2、微服务集成Seata

（1）引入依赖

<dependency>
    <groupId>com.alibaba.cloud</groupId>
    <artifactId>spring-cloud-starter-alibaba-seata</artifactId>
    <exclusions>
        <!--版本较低，1.3.0，因此排除-->
        <exclusion>
            <artifactId>seata-spring-boot-starter</artifactId>
            <groupId>io.seata</groupId>
        </exclusion>
    </exclusions>
</dependency>
<!--seata starter 采用1.4.2版本-->
<dependency>
    <groupId>io.seata</groupId>
    <artifactId>seata-spring-boot-starter</artifactId>
    <version>${seata.version}</version>
</dependency>

（2）修改application.yml

seata:
  registry: # TC服务注册中心的配置，微服务根据这些信息去注册中心获取tc服务地址
    # 参考tc服务自己的registry.conf中的配置
    type: nacos
    nacos: # tc
      server-addr: 127.0.0.1:8848
      namespace: ""
      group: DEFAULT_GROUP
      application: seata-tc-server # tc服务在nacos中的服务名称
      cluster: HZ
      username: nacos
      password: nacos
  tx-service-group: seata-demo # 事务组，根据这个获取tc服务的cluster名称
  service:
    vgroup-mapping: # 事务组与TC服务cluster的映射关系
      seata-demo: HZ

让微服务通过注册中心找到seata-tc-server （namespace+group+serviceName+cluster）

3、使用XA模式

（1）修改application.yml配置文件 

seata:
  data-source-proxy-mode: XA # 开启数据源代理的XA模式

（2）给发起全局事务的入口方法添加@GlobalTransactional注解

    @Override
    //@Transactional
    @GlobalTransactional
    public Long create(Order order) {
        // 创建订单
        orderMapper.insert(order);
        try {
            // 扣用户余额
            accountClient.deduct(order.getUserId(), order.getMoney());
            // 扣库存
            storageClient.deduct(order.getCommodityCode(), order.getCount());

        } catch (FeignException e) {
            log.error("下单失败，原因:{}", e.contentUTF8(), e);
            throw new RuntimeException(e.contentUTF8(), e);
        }
        return order.getId();
    }

4、使用AT模式(undo-log)

（1）全局锁锁表

关键记录： xid（事务id）、table_name（表名）、pk（锁的行id）

仅受Seata管理的事务才会被会局锁表管理

-- ----------------------------
-- Table structure for lock_table
-- ----------------------------
DROP TABLE IF EXISTS `lock_table`;
CREATE TABLE `lock_table`  (
  `row_key` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `xid` varchar(96) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `transaction_id` bigint(20) NULL DEFAULT NULL,
  `branch_id` bigint(20) NOT NULL,
  `resource_id` varchar(256) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `table_name` varchar(32) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `pk` varchar(36) CHARACTER SET utf8 COLLATE utf8_general_ci NULL DEFAULT NULL,
  `gmt_create` datetime NULL DEFAULT NULL,
  `gmt_modified` datetime NULL DEFAULT NULL,
  PRIMARY KEY (`row_key`) USING BTREE,
  INDEX `idx_branch_id`(`branch_id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci ROW_FORMAT = Compact;

（2）undo_log表

该表给微服务使用。

-- ----------------------------
-- Table structure for undo_log
-- ----------------------------
DROP TABLE IF EXISTS `undo_log`;
CREATE TABLE `undo_log`  (
  `branch_id` bigint(20) NOT NULL COMMENT 'branch transaction id',
  `xid` varchar(100) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'global transaction id',
  `context` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL COMMENT 'undo_log context,such as serialization',
  `rollback_info` longblob NOT NULL COMMENT 'rollback info',
  `log_status` int(11) NOT NULL COMMENT '0:normal status,1:defense status',
  `log_created` datetime(6) NOT NULL COMMENT 'create datetime',
  `log_modified` datetime(6) NOT NULL COMMENT 'modify datetime',
  UNIQUE INDEX `ux_undo_log`(`xid`, `branch_id`) USING BTREE
) ENGINE = InnoDB CHARACTER SET = utf8 COLLATE = utf8_general_ci COMMENT = 'AT transaction mode undo table' ROW_FORMAT = Compact;

-- ----------------------------
-- Records of undo_log
-- ----------------------------

（3）修改application.yml

seata:
  data-source-proxy-mode: AT # 开启数据源代理的AT模式

5、tcc模式(try \ confirm \ cancel)

（1）接口上使用@LocalTCC注解

（2）Try方法上使用@TwoPhaseBusinessAction注解，配置好confirm、cancel方法

（3）需要传递的参数使用@BusinessActionContextParameter注解配置。

（4）Confirm、Cancel方法需要boolean返回值。

@LocalTCC
public interface AccountTCCService {

    @TwoPhaseBusinessAction(name = "deduct", commitMethod = "confirm", rollbackMethod = "cancel")
    public void deduct(
            @BusinessActionContextParameter(paramName = "userId") String userId,
            @BusinessActionContextParameter(paramName = "money") int money
    );

    public boolean confirm(BusinessActionContext context);

    public boolean cancel(BusinessActionContext context);
}

（5）编写实现类

@Slf4j
@Service
public class AccountTCCServiceImpl implements AccountTCCService{

    @Autowired
    private AccountMapper accountMapper;
    @Autowired
    private AccountFreezeMapper accountFreezeMapper;

    @Override
    @Transactional
    public void deduct(String userId, int money) {
        // 0. 获取事务id
        String xid = RootContext.getXID();

        // 【悬挂操作】判断freeze中是否有冻结记录，如果有，一定是CANCEL执行过，需要拒绝业务
        AccountFreeze originFreeze = accountFreezeMapper.selectById(xid);
        if (originFreeze != null){
            // CANCEL执行过，需要拒绝业务
            return;
        }

        // 1. 扣减可用余额
        accountMapper.deduct(userId, money);
        // 2. 记录冻结金额，事务状态
        // 2.1 设置冻结金额信息
        AccountFreeze freeze = new AccountFreeze();
        freeze.setUserId(userId);
        freeze.setFreezeMoney(money);
        freeze.setState(AccountFreeze.State.TRY);
        freeze.setXid(xid);
        // 2.2 将信息写入数据库
        accountFreezeMapper.insert(freeze);
    }


    @Override
    public boolean confirm(BusinessActionContext context) {
        // 1. 获取事务id
        String xid = context.getXid();
        // 2. 根据事务id删除冻结金额
        int count = accountFreezeMapper.deleteById(xid);
        return count == 1;
    }

    @Override
    public boolean cancel(BusinessActionContext context) {
        // 0. 查询冻结记录
        String xid = context.getXid();
        String userId = context.getActionContext("userId").toString();
        AccountFreeze freeze = accountFreezeMapper.selectById(xid);

        // 【空回滚】的判断，判断freeze对象是否为null，为null证明try没有执行，需要空回滚
        if (freeze == null){
            freeze = new AccountFreeze();
            freeze.setUserId(userId);
            freeze.setFreezeMoney(0);
            freeze.setState(AccountFreeze.State.CANCEL);
            freeze.setXid(xid);
            accountFreezeMapper.insert(freeze);
            return true;
        }

        // 【幂等】判断
        if (freeze.getState() == AccountFreeze.State.CANCEL){
            // 已经处理过一次CANCEL了，无需重复处理
            return true;
        }

        // 1. 恢复可用金额
        accountMapper.refund(freeze.getUserId(), freeze.getFreezeMoney());

        // 2. 将冻结金额清零，状态改为CANCEL
        freeze.setFreezeMoney(0);
        freeze.setState(AccountFreeze.State.CANCEL);
        int count = accountFreezeMapper.updateById(freeze);
        return count == 1;
    }
}

（6）表格准备

CREATE TABLE `account_tbl` (
  `id` int NOT NULL AUTO_INCREMENT,
  `user_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL,
  `money` int unsigned DEFAULT '0',
  PRIMARY KEY (`id`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=2 DEFAULT CHARSET=utf8mb3 ROW_FORMAT=COMPACT;
CREATE TABLE `account_freeze_tbl` (
  `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `user_id` varchar(255) CHARACTER SET utf8 COLLATE utf8_general_ci DEFAULT NULL,
  `freeze_money` int unsigned DEFAULT '0',
  `state` int DEFAULT NULL COMMENT '事务状态，0:try，1:confirm，2:cancel',
  PRIMARY KEY (`xid`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 ROW_FORMAT=COMPACT;

 

 

 

相关推荐
Mybatis Cause: java.lang.ClassNotFoundException: Cannot find class: BaseResultMap
原因很简单，就一句话，是不是resultType 和resultMap 弄混了？ &nbsp; 大致的意思就是无法找到返回值对应的r
eureka consul nacos对比（CAP理论）
一、CAP理论 英文全称Dynamic Naming and Configuration Service。在描述下面内容之前，先介绍一下CAP原则：一个分布式系统不可能同时满足C和A和P这3个条件。
netty处理TCP粘包 DelimiterBasedFrameDecoder
LineBasedFrameDecoder以换行符\n或\r\n作为依据，遇到\r\n都认为是一条完整的消息。 而DelimiterBasedFrameDecoder允许我们指定任
mybatis调用链路
一、概述 二、SqlSessionFactory SqlSessionFactory由Mybatis配置文件和映射文件，使用SqlSessionFactoryBuilder进行创建，它是用于获取Sq
spring cloud alibaba
一、概述 第一代的spring cloud标准中很多组件已停更，如eureak、zuul等，Spring Cloud Alibaba是阿里的微服务的解决方案。依托Spring Cloud Alibab
flowable liquibase
一、概述 Flowable引擎使用Liquibase来控制其数据库表的版本，如果使用spring boot整合flowable时，这意味着LiquibaseAutoConfiguration来自动启动
java中使用base64
一、Base64 Base64是一种基于64（8Bit）个可打印字符来表示二进制数据的方法。 索引 对应字符 索引 对应字符 索引 对应字符 索引 对应字符
生成验证码工具easy-captcha
一、概述 https://github.com/pig-mesh/easy-captcha easy-captcha是生成图形验证码的Java类库，支持gif、中文、算术等类型。 效果： 二、使用
netty LengthFieldBasedFrameDecoder 自定义长度帧解码器
一、概述 一个Netty的ByteBuf包，可以被TCP拆分成多个二进制数据帧进行发送，也有可能，底层将多个小的ByteBuf包，封装成一个大的底层数据帧发送出去。 如何从底层的二进制数据帧中，界定出