文章/答案/技术大牛

发布

社区首页 >专栏 >5 张图带你了解分布式事务 Saga 模式中的状态机

5 张图带你了解分布式事务 Saga 模式中的状态机

jinjunzhu

发布于 2024-03-20 12:36:33

73400

代码可运行

文章被收录于专栏：个人开发个人开发

运行总次数：0

代码可运行

大家好，我是君哥。

状态机在我们的工作中应用非常广泛，今天聊一聊分布式事务中间件 Seata 中 Saga 模式的状态机。

1 状态机简介

状态机是一个数学模型，它将工作中的运行状态和流转规则抽象出来，可以协调相关信号来完成预先设定的操作。

下面介绍状态机中的几个概念：

状态：状态机目前的状态标识；
状态转移：定义状态之间的转移路由；
动作（Action）：状态转移需要的操作；
事件：要执行某个操作时的触发器或者口令。

状态机一般用在状态类型比较多（超过 3 个），分支流程比较多，初始状态经过多个流程的流转达到最终状态的场景。

2 Saga 模式

Saga 模式是分布式事务中长事务的一种解决方案，Seata 中 Saga 模式的理论基础是 Hector & Kenneth 在 1987 年发表的论文 Sagas。下图（来自官网）是 Seata 中 Saga 模型：

在 Saga 模式中，如果一部分分支事务已经提交成功，当其中一个分支事务提交失败，状态机就会触发所有提交成功的分支事务进行回滚。

分支事务中提交和回滚的逻辑需要由业务代码来实现。

3 Saga 实现

Seata 中 Saga 模式是基于状态机来实现的，使用 Saga 模式时，先画一张状态图，这个状态图定义服务调用流程，每个节点调用一个分支事务，并且每个节点需要配备一个补偿节点用于分支事务失败后的补偿动作。

以经典电商案例来讲，一个分布式事务中有三个分支事务参数者：

分支事务	动作	状态
订单服务	保存订单	保存成功、失败
账户服务	扣减金额	扣减成功、失败
库存服务	扣减库存	扣减成功、失败

在这个分布式事务中，只有订单、账户、库存这三个分支事务都提交成功，整个事务才能成功。每一个分支事务提交失败，其他执行成功的事务都需要反向补偿。如下图：

比如扣减金额这个分支事务失败了，需要反向补偿扣减金额、保存订单这两个分支事务。那 Seata 是怎么做到事件触发、状态流转和补偿操作的呢？

使用 Seata 状态机，首先需要定义一个 Json 文件，这个 Json 文件把图中的每个节点都定义成一个 State，State 的类型共有四种：

ServiceTask：对应分支事务的提交操作；
Choice：对应流程中下一个 State 的选择；
CompensationTrigger：触发补偿服务；
Succeed：成功状态，当所有分支事务都成功后才会流转到这个状态；
Fail：失败状态。

3.1 ServiceTask

下面我们看"保存订单"这个状态：

"SaveOrder": {
 "Type": "ServiceTask",
 "ServiceName": "orderSave",
 "ServiceMethod": "saveOrder",
 "CompensateState": "DeleteOrder",
 "Next": "ChoiceAccountState",
 "Input": [
  "$.[businessKey]",
  "$.[order]"
 ],
 "Output": {
  "SaveOrderResult": "$.#root"
 },
 "Status": {
  "#root == true": "SU",
  "#root == false": "FA",
  "$Exception{java.lang.Throwable}": "UN"
 },
 "Catch": [
  {
   "Exceptions": [
    "java.lang.Throwable"
   ],
   "Next": "CompensationTrigger"
  }
 ]
},

这个 State 的类型是 ServiceTask，上面图中的分支服务和补偿服务都是这种类型，也对应代码中的一个 Service。上面的 Json 中主要定义了三个内容：

这个 state 调用的 Service 方法；
提交失败后的补偿 State(CompensateState)；
提交成功后应该跳转的下一个 State(ChoiceAccountState)。

3.2 Choice

下面来看 ChoiceAccountState 这个状态节点，Json 文件定义如下：

"ChoiceAccountState":{
 "Type": "Choice",
 "Choices":[
  {
   "Expression":"[SaveOrderResult] == true",
   "Next":"ReduceAccount"
  }
 ],
 "Default":"Fail"
}

对应的下个节点是 ReduceAccount，如果失败就会跳转 Fail 状态。

3.3 Fail

上面 orderSave 这个状态节点如果发生异常，会跳转到 CompensationTrigger，CompensationTrigger 状态节点定义如下：

"CompensationTrigger": {
 "Type": "CompensationTrigger",
 "Next": "Fail"
}

这个节点会触发 SaveOrder 中定义的补偿服务，然后将最终状态流转到 Fail。同时我们也看到，只要到了 CompensationTrigger 这个状态节点，最终状态就会流转到 Fail。

下面我们把整个 Json 文件的定义贴出来看一下：

{
    "Name": "buyGoodsOnline",
    "Comment": "buy a goods on line, add order, deduct account, deduct storage ",
    "StartState": "SaveOrder",
    "Version": "0.0.1",
 #定义状态
    "States": {
        "SaveOrder": {
            "Type": "ServiceTask",
            "ServiceName": "orderSave",
            "ServiceMethod": "saveOrder",
            "CompensateState": "DeleteOrder",
            "Next": "ChoiceAccountState",
            "Input": [
                "$.[businessKey]",
                "$.[order]"
            ],
            "Output": {
                "SaveOrderResult": "$.#root"
            },
            "Status": {
                "#root == true": "SU",
                "#root == false": "FA",
                "$Exception{java.lang.Throwable}": "UN"
            },
   "Catch": [
                {
                    "Exceptions": [
                        "java.lang.Throwable"
                    ],
                    "Next": "CompensationTrigger"
                }
            ]
        },
        "ChoiceAccountState":{
            "Type": "Choice",
            "Choices":[
                {
                    "Expression":"[SaveOrderResult] == true",
                    "Next":"ReduceAccount"
                }
            ],
            "Default":"Fail"
        },
        "ReduceAccount": {
            "Type": "ServiceTask",
            "ServiceName": "accountService",
            "ServiceMethod": "decrease",
            "CompensateState": "CompensateReduceAccount",
            "Next": "ChoiceStorageState",
            "Input": [
                "$.[businessKey]",
                "$.[userId]",
                "$.[money]",
                {
                    "throwException" : "$.[mockReduceAccountFail]"
                }
            ],
            "Output": {
                "ReduceAccountResult": "$.#root"
            },
            "Status": {
                "#root == true": "SU",
                "#root == false": "FA",
                "$Exception{java.lang.Throwable}": "UN"
            },
            "Catch": [
                {
                    "Exceptions": [
                        "java.lang.Throwable"
                    ],
                    "Next": "CompensationTrigger"
                }
            ]
        },
        "ChoiceStorageState":{
            "Type": "Choice",
            "Choices":[
                {
                    "Expression":"[ReduceAccountResult] == true",
                    "Next":"ReduceStorage"
                }
            ],
            "Default":"Fail"
        },
        "ReduceStorage": {
            "Type": "ServiceTask",
            "ServiceName": "storageService",
            "ServiceMethod": "decrease",
            "CompensateState": "CompensateReduceStorage",
            "Input": [
                "$.[businessKey]",
                "$.[productId]",
                "$.[count]",
                {
                    "throwException" : "$.[mockReduceStorageFail]"
                }
            ],
            "Output": {
                "ReduceStorageResult": "$.#root"
            },
            "Status": {
                "#root == true": "SU",
                "#root == false": "FA",
                "$Exception{java.lang.Throwable}": "UN"
            },
            "Catch": [
                {
                    "Exceptions": [
                        "java.lang.Throwable"
                    ],
                    "Next": "CompensationTrigger"
                }
            ],
            "Next": "Succeed"
        },
        "DeleteOrder": {
            "Type": "ServiceTask",
            "ServiceName": "orderSave",
            "ServiceMethod": "deleteOrder",
            "Input": [
                "$.[businessKey]",
                "$.[order]"
            ]
        },
        "CompensateReduceAccount": {
            "Type": "ServiceTask",
            "ServiceName": "accountService",
            "ServiceMethod": "compensateDecrease",
            "Input": [
                "$.[businessKey]",
                "$.[userId]",
                "$.[money]"
            ]
        },
        "CompensateReduceStorage": {
            "Type": "ServiceTask",
            "ServiceName": "storageService",
            "ServiceMethod": "compensateDecrease",
            "Input": [
                "$.[businessKey]",
                "$.[productId]",
                "$.[count]"
            ]
        },
        "CompensationTrigger": {
            "Type": "CompensationTrigger",
            "Next": "Fail"
        },
        "Succeed": {
            "Type":"Succeed"
        },
        "Fail": {
            "Type":"Fail",
            "ErrorCode": "PURCHASE_FAILED",
            "Message": "purchase failed"
        }
    }
}

上面 Json 文件中定义的 buyGoodsOnline，是状态机加载的入口，状态机会找到这个 name，然后把状态加载到自己的内存中。下面，我们再来总结一下电商案例中分布式事务状态流转过程：

4 状态机应用

上面的电商例子中，三个分支服务分别定义了三个 State，对应的 ServiceMethod 如下：

SaveOrder#saveOrder：

public boolean saveOrder(String businessKey, Order order) {
 logger.info("保存订单, businessKey：{}, order: {}", businessKey, order);
 orderDao.create(order);
 return true;
}

ReduceAccount#decrease

public boolean decrease(String businessKey, Long userId, BigDecimal money) {
 return accountApi.decrease(businessKey, userId, money);
}

ReduceStorage#decrease

public boolean decrease(String businessKey, Long productId, Integer count) {
 return storageApi.decrease(businessKey, productId, count);
}

状态机在启动的时候，需要把上面方法中的参数都传入，实例代码如下：

StateMachineEngine stateMachineEngine = (StateMachineEngine) ApplicationContextUtils.getApplicationContext().getBean("stateMachineEngine");
Map<String, Object> startParams = new HashMap<>(3);
String businessKey = String.valueOf(System.currentTimeMillis());
startParams.put("businessKey", businessKey);
startParams.put("order", order);
startParams.put("mockReduceAccountFail", "true");
startParams.put("userId", order.getUserId());
startParams.put("money", order.getPayAmount());
startParams.put("productId", order.getProductId());
startParams.put("count", order.getCount());
//这里采用同步方法
StateMachineInstance inst = stateMachineEngine.startWithBusinessKey("buyGoodsOnline", null, businessKey, startParams);

5 状态机原理

下面这张图来自于 Seata 官网，主要讲解了状态机的工作原理：

状态机启动时，首先启动了全局事务；
将状态机的参数记录在本地 seata_state_machine_inst 表；
向 Seata Server 注册分支事务；
执行 StateA 并记录状态到本地数据库，同时会产生路由事件放入 EventQueue，执行 StateB 时取出路由消息触发执行。同样 StateB 执行时也会产生路由消息放入 EventQueue；
从 EventQueue 取出路由消息执行 StateC；
状态机结束流程，提交或回滚全局事务。