微服务中间件--分布式事务_微服务的中间件

a.理论基础

1) CAP定理

分布式系统有三个指标：

• Consistency（一致性）: 用户访问分布式系统中的任意节点，得到的数据必须一致
• Availability（可用性）: 用户访问集群中的任意健康节点，必须能得到响应，而不是超时或拒绝
• Partition tolerance （分区容错性）
  • Partition(分区): 因为网络故障或其它原因导致分布式系统中的部分节点与其它节点失去连接，形成独立分区
  • tolerance(容错): 在集群出现分区时，整个系统也要持续对外提供服务

分布式系统无法同时满足这三个指标，这个结论就叫做 CAP 定理。

2) BASE理论

BASE理论是对CAP的一种解决思路，包含三个思想：

• Basically Available （基本可用）：分布式系统在出现故障时，允许损失部分可用性，即保证核心可用
• **Soft State（软状态）：**在一定时间内，允许出现中间状态，比如临时的不一致状态
• **Eventually Consistent（最终一致性）：**虽然无法保证强一致性，但是在软状态结束后，最终达到数据一致

而分布式事务最大的问题是各个子事务的一致性问题，因此可以借鉴CAP定理和BASE理论：

• AP模式：各子事务分别执行和提交，允许出现结果不一致，然后采用弥补措施恢复数据即可，实现最终一致
• CP模式：各个子事务执行后互相等待，同时提交，同时回滚，达成强一致。但事务等待过程中，处于弱可用状态。

b.Seata

Seata是 2019 年 1 月份蚂蚁金服和阿里巴巴共同开源的分布式事务解决方案。致力于提供高性能和简单易用的分布式事务服务，为用户打造一站式的分布式解决方案。官网地址：http://seata.io/，其中的文档、播客中提供了大量的使用说明、源码分析。

Seata架构

Seata事务管理中有三个重要的角色：

• TC (Transaction Coordinator) - 事务协调者：维护全局和分支事务的状态，协调全局事务提交或回滚
• TM (Transaction Manager) - 事务管理器：定义全局事务的范围、开始全局事务、提交或回滚全局事务
• RM (Resource Manager) - 资源管理器：管理分支事务处理的资源，与TC交谈以注册分支事务和报告分支事务的状态，并驱动分支事务提交或回滚

Seata提供了四种不同的分布式事务解决方案：

• XA模式：强一致性分阶段事务模式，牺牲了一定的可用性，无业务侵入
• TCC模式：最终一致的分阶段事务模式，有业务侵入
• AT模式：最终一致的分阶段事务模式，无业务侵入，也是Seata的默认模式
• SAGA模式：长事务模式，有业务侵入

1) XA模式

XA模式原理

XA 规范 是 X/Open 组织定义的分布式事务处理（DTP，Distributed Transaction Processing）标准，XA 规范 描述了全局的TM与局部的RM之间的接口，几乎所有主流的数据库都对 XA 规范 提供了支持。

seata的XA模式：seata的XA模式做了一些调整，但大体相似：

RM一阶段的工作：

• 1.注册分支事务到TC
• 2.执行分支业务sql，但不提交
• 3.报告执行状态到TC

TC二阶段的工作：

• TC检测各分支事务执行状态
  • a.如果都成功，通知所有RM提交事务
  • b.如果有失败，通知所有RM回滚事务

RM二阶段的工作：

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

XA模式的优点是什么？

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

XA模式的缺点是什么？

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

1.a) 实现XA模式

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

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

seata:
	data-source-proxy-mode: XA
1
2

2.给发起全局事务的入口方法添加@GlobalTransactional注解，本例中是OrderServiceImpl中的create方法：

@Override
@GlobalTransactional
public Long create(Order order) {
    // 创建订单
    orderMapper.insert(order);
    // 扣余额 ...略 
    // 扣减库存 ...略
    return order.getId();
}
1
2
3
4
5
6
7
8
9

3.重启服务并测试

2) AT模式

AT模式原理

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

阶段一RM的工作：

• 1.注册分支事务
• 2.记录undo-log（数据快照）
• 3.执行业务sql并提交
• 4.报告事务状态

阶段二提交时RM的工作：

• 删除undo-log即可

阶段二回滚时RM的工作：

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

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

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

AT模式的存在脏写问题

AT模式的优点：

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

AT模式的缺点：

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

3) TCC模式

TCC模式原理

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

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

TCC的优点是什么？

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

TCC的缺点是什么？

• 有代码侵入，需要人为编写try、Confirm和Cancel接口，太麻烦
• 软状态，事务是最终一致
• 需要考虑Confirm和Cancel的失败情况，做好幂等处理

3.a) 代码实现

案例：改造account-service服务，利用TCC实现分布式事务

需求如下：

• 修改account-service，编写try、confirm、cancel逻辑
• try业务：添加冻结金额，扣减可用金额
• confirm业务：删除冻结金额
• cancel业务：删除冻结金额，恢复可用金额
• 保证confirm、cancel接口的幂等性
• 允许空回滚
• 拒绝业务悬挂

TCC的空回滚和业务悬挂

当某分支事务的try阶段阻塞时，可能导致全局事务超时而触发二阶段的cancel操作。在未执行try操作时先执行了cancel操作，这时cancel不能做回滚，就是空回滚。

对于已经空回滚的业务，如果以后继续执行try，就永远不可能confirm或cancel，这就是业务悬挂。应当阻止执行空回滚后的try操作，避免悬挂

声明TCC接口

TCC的Try、Confirm、Cancel方法都需要在接口中基于注解来声明，语法如下：

在Account-service中的service下定义接口

@LocalTCC
public interface AccountTCCService {

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

    boolean confirm(BusinessActionContext ctx);

    boolean cancel(BusinessActionContext ctx);
}
1
2
3
4
5
6
7
8
9
10
11

在Account-service中的service的impl下实现该方法

@Service
@Slf4j
public class AccountTCCServiceImpl implements AccountTCCService {

    @Autowired
    private AccountMapper accountMapper;

    @Autowired
    private AccountFreezeMapper freezeMapper;

    @Override
    @Transactional
    public void deduct(String userId, int money) {
        // 0.获取事务id
        String xid = RootContext.getXID();
        // 1.判断freeze中是否有冻结金额，如果有，一定是CANCEL执行过，要拒绝业务
        AccountFreeze oldFreeze = freezeMapper.selectById(userId);
        if (oldFreeze != null){
            // CANCEL执行过，要拒绝业务
            return;
        }
        // 1.扣减可用余额
        accountMapper.deduct(userId, money);
        // 2.记录冻结金额，记录事务状态
        AccountFreeze freeze = new AccountFreeze();
        freeze.setUserId(userId);
        freeze.setFreezeMoney(money);
        freeze.setState(AccountFreeze.State.TRY);
        freeze.setXid(xid);
        freezeMapper.insert(freeze);
    }

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

    @Override
    public boolean cancel(BusinessActionContext ctx) {
        // 0.查询冻结记录
        String xid = ctx.getXid();
        String userId = ctx.getActionContext("userId").toString();
        AccountFreeze freeze = freezeMapper.selectById(xid);
        // 1.空回滚的判断，判断freeze是否为null，为Null证明try没执行，需要空回滚
        if (freeze == null){
            // 证明try没执行，需要空回滚
            freeze = new AccountFreeze();
            freeze.setUserId(userId);
            freeze.setFreezeMoney(0);
            freeze.setState(AccountFreeze.State.CANCEL);
            freeze.setXid(xid);
            freezeMapper.insert(freeze);
        }
        // 2.幂等判断
        if (freeze.getState().equals(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 = freezeMapper.updateById(freeze);

        return count == 1;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73

4) Saga模式

Saga模式是Seata提供的长事务解决方案。也分为两个阶段：

• 一阶段：直接提交本地事务
• 二阶段：成功则什么都不做；失败则通过编写补偿业务来回滚

Saga模式优点：

• 事务参与者可以基于事件驱动实现异步调用，吞吐高
• 一阶段直接提交事务，无锁，性能好
• 不用编写TCC中的三个阶段，实现简单

缺点：

• 软状态持续时间不确定，时效性差
• 没有锁，没有事务隔离，会有脏写

5) 四种模式对比

6) TC的异地多机房容灾架构

TC服务作为Seata的核心服务，一定要保证高可用和异地容灾。

现在，将seata目录复制一份，起名为seata2

1.修改seata2/conf/registry.conf内容如下：

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"
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

进入seata2/bin目录，然后运行命令：

seata-server.bat -p 8092
1

打开nacos控制台，查看服务列表：

点进详情查看：

2.将事务组映射配置到nacos

接下来，我们需要将tx-service-group与cluster的映射关系都配置到nacos配置中心。

新建一个配置：

配置的内容如下：

# 事务组映射关系
service.vgroupMapping.seata-demo=SH

service.enableDegrade=false
service.disableGlobalTransaction=false
# 与TC服务的通信配置
transport.type=TCP
transport.server=NIO
transport.heartbeat=true
transport.enableClientBatchSendRequest=false
transport.threadFactory.bossThreadPrefix=NettyBoss
transport.threadFactory.workerThreadPrefix=NettyServerNIOWorker
transport.threadFactory.serverExecutorThreadPrefix=NettyServerBizHandler
transport.threadFactory.shareBossWorker=false
transport.threadFactory.clientSelectorThreadPrefix=NettyClientSelector
transport.threadFactory.clientSelectorThreadSize=1
transport.threadFactory.clientWorkerThreadPrefix=NettyClientWorkerThread
transport.threadFactory.bossThreadSize=1
transport.threadFactory.workerThreadSize=default
transport.shutdown.wait=3
# RM配置
client.rm.asyncCommitBufferLimit=10000
client.rm.lock.retryInterval=10
client.rm.lock.retryTimes=30
client.rm.lock.retryPolicyBranchRollbackOnConflict=true
client.rm.reportRetryCount=5
client.rm.tableMetaCheckEnable=false
client.rm.tableMetaCheckerInterval=60000
client.rm.sqlParserType=druid
client.rm.reportSuccessEnable=false
client.rm.sagaBranchRegisterEnable=false
# TM配置
client.tm.commitRetryCount=5
client.tm.rollbackRetryCount=5
client.tm.defaultGlobalTransactionTimeout=60000
client.tm.degradeCheck=false
client.tm.degradeCheckAllowTimes=10
client.tm.degradeCheckPeriod=2000

# undo日志配置
client.undo.dataValidation=true
client.undo.logSerialization=jackson
client.undo.onlyCareUpdateColumns=true
client.undo.logTable=undo_log
client.undo.compress.enable=true
client.undo.compress.type=zip
client.undo.compress.threshold=64k
client.log.exceptionRate=100
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

3.微服务读取nacos配置

接下来，需要修改每一个微服务的application.yml文件，让微服务读取nacos中的client.properties文件：

seata:
  config:
    type: nacos
    nacos:
      server-addr: 127.0.0.1:8848
      username: nacos
      password: nacos
      group: SEATA_GROUP
      data-id: client.properties
1
2
3
4
5
6
7
8
9

重启微服务，现在微服务到底是连接tc的SH集群，还是tc的HZ集群，都统一由nacos的client.properties来决定了。