微服务系列文章之 seata 事务模式-Toy模板网

这篇具有很好参考价值的文章主要介绍了微服务系列文章之 seata 事务模式。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

XA模式

XA 规范是 X/Open 组织定义的分布式事务处理（DTP，Distributed Transaction Processing）标准。

XA 规范描述了全局的TM与局部的RM之间的接口，几乎所有主流的数据库都对 XA 规范提供了支持。

两阶段提交

XA是规范，目前主流数据库都实现了这种规范，实现的原理都是基于两阶段提交。

正常情况：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

异常情况：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

一阶段：

事务协调者通知每个事物参与者（分支事务）执行本地事务
本地事务执行完成后报告事务执行状态给事务协调者，此时事务不提交，继续持有数据库锁（行锁）

二阶段：

事务协调者基于一阶段的报告来判断下一步操作
- 如果一阶段都成功，则通知所有事务参与者，提交事务
- 如果一阶段任意一个参与者失败，则通知所有事务参与者回滚事务

Seata的XA模型

Seata对原始的XA模式做了简单的封装和改造，以适应自己的事务模型，基本架构如图：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

RM一阶段的工作：

① 注册分支事务到TC

② 执行分支业务sql但不提交

③ 报告执行状态到TC

TC二阶段的工作：

TC检测各分支事务执行状态

a.如果都成功，通知所有RM提交事务

b.如果有失败，通知所有RM回滚事务

RM二阶段的工作：

接收TC指令，提交或回滚事务

优缺点

XA模式的优点是什么？

事务的强一致性，满足ACID原则。
常用（关系型）数据库都支持，实现简单，并且没有代码侵入

XA模式的缺点是什么？

因为一阶段需要锁定数据库资源（行锁），等待二阶段结束才释放，性能较差
依赖关系型数据库实现事务（redis不支持）

实现XA模式

Seata的starter已经完成了XA模式的自动装配，实现非常简单，步骤如下：

1）修改application.yml文件（每个参与事务的微服务），开启XA模式：

seata:
  data-source-proxy-mode: XA

微服务系列文章之 seata 事务模式,微服务,架构,云原生

2）给发起全局事务的入口方法添加@GlobalTransactional注解:
微服务系列文章之 seata 事务模式,微服务,架构,云原生

3）重启服务并测试

重启order-service，再次测试，发现无论怎样，三个微服务都能成功回滚。

AT模式

AT模式同样是分阶段提交的事务模型，不过缺弥补了XA模型中资源锁定周期过长的缺陷。

Seata的AT模型

微服务系列文章之 seata 事务模式,微服务,架构,云原生

阶段一RM的工作：

注册分支事务
记录undo-log（数据快照）
执行业务sql并提交（不存在行锁）
报告事务状态

阶段二提交（成功）时RM的工作：

删除undo-log即可

阶段二回滚（失败）时RM的工作：

根据undo-log恢复数据到更新前

流程梳理

例如，一个分支业务的SQL是这样的：update tb_account set money = money - 10 where id = 1

微服务系列文章之 seata 事务模式,微服务,架构,云原生

AT与XA的区别

简述AT模式与XA模式最大的区别是什么？

XA模式一阶段不提交事务，锁定资源；AT模式一阶段直接提交，不锁定资源。
XA模式依赖数据库机制实现回滚；AT模式利用数据快照实现数据回滚。
XA模式强一致；AT模式最终一致

脏写问题

在多线程并发访问AT模式的分布式事务时，有可能出现脏写问题，如图：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

解决思路就是引入了Seata全局锁的概念。在释放DB锁之前，先拿到全局锁。避免同一时刻有另外一个事务来操作当前数据。

微服务系列文章之 seata 事务模式,微服务,架构,云原生

优缺点

AT模式的优点：

一阶段完成直接提交事务，释放数据库资源，性能比较好
利用全局锁实现读写隔离
没有代码侵入，框架自动完成回滚和提交

AT模式的缺点：

两阶段之间属于软状态，属于最终一致
框架的快照功能会影响性能，但比XA模式要好很多

实现AT模式

AT模式中的快照生成、回滚等动作都是由框架自动完成，没有任何代码侵入，因此实现非常简单。

只不过，AT模式需要一个表来记录全局锁、另一张表来记录数据快照undo_log。

导入数据库表

seata-tc-server内部记录全局锁，每个微服务中需要存undo_log

微服务系列文章之 seata 事务模式,微服务,架构,云原生

修改application.yml文件

将事务模式修改为AT模式即可：

seata:
  data-source-proxy-mode: AT # 默认就是AT

重启服务并测试

TCC模式

TCC模式与AT模式非常相似，每阶段都是独立事务，不同的是TCC通过人工编码来实现数据恢复。需要实现三个方法：

Try：资源的检测和预留；
Confirm：完成资源操作业务；要求 Try 成功 Confirm 一定要能成功。
Cancel：预留资源释放，可以理解为try的反向操作。

TCC模式原理

举例，一个扣减用户余额的业务。假设账户A原来余额是100，需要余额扣减30元。

阶段一（ Try ）：检查余额是否充足，如果充足则冻结金额增加30元，可用余额扣除30

微服务系列文章之 seata 事务模式,微服务,架构,云原生

阶段二（Confirm)：假如要提交（Confirm），则冻结金额扣减30

确认可以提交，不过之前可用金额已经扣减过了，这里只要清除冻结金额就好了：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

阶段二(Canncel)：如果要回滚（Cancel），则冻结金额扣减30，可用余额增加30

需要回滚，那么就要释放冻结金额，恢复可用金额：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

Seata的TCC模型

Seata中的TCC模型依然延续之前的事务架构，如图：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

优缺点

TCC模式的每个阶段是做什么的？

Try：资源检查和预留
Confirm：业务执行和提交
Cancel：预留资源的释放

TCC的优点是什么？

一阶段完成直接提交事务，释放数据库资源，性能好
相比AT模型，无需生成快照，无需使用全局锁，性能最强
不依赖数据库事务，而是依赖补偿操作，可以用于非事务型数据库（redis）

TCC的缺点是什么？

有代码侵入，需要人为编写try、Confirm和Cancel接口，太麻烦
软状态，事务是最终一致
需要考虑Confirm和Cancel的失败情况，做好幂等处理（发一次请求执行结果，跟发10请求执行结果要求一致）

业务悬挂和空回滚

空回滚

当某分支事务的try阶段阻塞时，可能导致全局事务超时而触发二阶段的cancel操作。

在未执行try操作时先执行了cancel操作，这时cancel不能做回滚，就是空回滚。

微服务系列文章之 seata 事务模式,微服务,架构,云原生

执行cancel操作时，应当判断try是否已经执行，如果尚未执行，则应该空回滚。

业务悬挂

对于已经空回滚的业务，之前被阻塞的try操作恢复，继续执行try，就永远不可能confirm或cancel ，事务一直处于中间状态，这就是业务悬挂。

执行try操作时，应当判断cancel是否已经执行过了，如果已经执行，应当阻止空回滚后的try操作，避免悬挂

微服务系列文章之 seata 事务模式,微服务,架构,云原生

实现TCC模式

业务分析

为了实现空回滚、防止业务悬挂，以及幂等性要求。我们必须在数据库记录冻结金额的同时，记录当前事务id和执行状态，为此我们设计了一张表：

微服务系列文章之 seata 事务模式,微服务,架构,云原生

准备TCC环境

1）创建表

CREATE TABLE `tb_account_freeze`  (
  `xid` varchar(128) CHARACTER SET utf8 COLLATE utf8_general_ci NOT NULL,
  `user_id` int(11) NOT NULL,
  `freeze_money` int(11) UNSIGNED NULL DEFAULT 0,
  `state` int(1) NULL DEFAULT NULL COMMENT '事务状态，0:try，1:confirm，2:cancel',  
  PRIMARY KEY (`xid`) USING BTREE
);

2）创建实体

@Data
@TableName("tb_account_freeze")
public class AccountFreeze {
    @TableId(type = IdType.INPUT)
    private String xid;
    private Integer userId;
    private Integer freezeMoney;
    private Integer state;

    public static abstract class State {
        public final static int TRY = 0;
        public final static int CONFIRM = 1;
        public final static int CANCEL = 2;
    }
}

3）AccountFreezeMapper

public interface AccountFreezeMapper extends BaseMapper<AccountFreeze> {
}

声明TCC接口

@LocalTCC
public interface AccountTCCService {

    /**
     *  一阶段：try 预留
     *  @TwoPhaseBusinessAction 定义 try confirm cancel 方法名
     */
    @TwoPhaseBusinessAction(name = "deduct", commitMethod = "confirm", rollbackMethod = "cancel")
    void deduct(@BusinessActionContextParameter(paramName = "userId") Integer userId,
                @BusinessActionContextParameter(paramName = "money") Integer money);

    /**
     *  二阶段：confirm 提交
     * @param ctx 可以获得 @BusinessActionContextParameter参数和xid
     * 
     */
    Boolean confirm(BusinessActionContext ctx);

    /**
     * 二阶段：cancel 回滚
     * @param ctx 可以获得 @BusinessActionContextParameter参数和xid
     *  
     */
    Boolean cancel(BusinessActionContext ctx);
}

实现TCC接口

package com.heima.account.service.impl;

import com.heima.account.entity.AccountFreeze;
import com.heima.account.mapper.AccountFreezeMapper;
import com.heima.account.mapper.AccountMapper;
import com.heima.account.service.AccountTCCService;
import com.sun.org.apache.regexp.internal.RE;
import io.seata.core.context.RootContext;
import io.seata.rm.tcc.api.BusinessActionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class AccountTCCServiceImpl implements AccountTCCService {

    @Autowired
    private AccountMapper accountMapper;

    @Autowired
    private AccountFreezeMapper freezeMapper;

    @Override
    public void deduct(Integer userId, Integer money) {
        // 0.获取事务id
        String xid = RootContext.getXID();
        // 解决业务悬挂
        AccountFreeze oldFreeze = freezeMapper.selectById(xid);
        if (oldFreeze != null) {
            return;
        }
        // 1.扣除可用余额
        accountMapper.deduct(userId, money);
        // 2.记录冻结金额，事务状态
        AccountFreeze freeze = new AccountFreeze();
        freeze.setXid(xid);
        freeze.setUserId(userId);
        freeze.setFreezeMoney(money);
        freeze.setState(AccountFreeze.State.TRY);
        freezeMapper.insert(freeze);

    }

    @Override
    public boolean confirm(BusinessActionContext ctx) {
        // 1.获取xid
        String xid = ctx.getXid();
        // 2.删除冻结记录
        int count = freezeMapper.deleteById(xid);
        return count == 1;
    }

    @Override
    @Transactional
    public boolean cancel(BusinessActionContext ctx) {
        // 1.查询冻结记录
        String xid = ctx.getXid();
        AccountFreeze freeze = freezeMapper.selectById(xid);

        // 解决空回滚
        if (freeze == null) {
            int userId = Integer.parseInt(ctx.getActionContext("userId").toString());
            freeze = new AccountFreeze();
            freeze.setXid(xid);
            freeze.setUserId(userId);
            freeze.setFreezeMoney(0);
            freeze.setState(AccountFreeze.State.CANCEL);
            freezeMapper.insert(freeze);
            return true;
        }

        // 解决幂等性
        if (freeze.getState() == AccountFreeze.State.CANCEL) {
            return true;
        }

        // 2.恢复账户金额
        accountMapper.refund(freeze.getUserId(), freeze.getFreezeMoney());
        // 3.更新冻结记录
        freeze.setFreezeMoney(0);
        freeze.setState(AccountFreeze.State.CANCEL);
        int count = freezeMapper.updateById(freeze);
        return count == 1;
    }
}