五張圖帶你了解分布式事務 Saga 模式中的狀態機

作者：朱晉君 2024-03-12 08:35:47

開發架構

本文講解了分布式事務中間件 Seata 給 Saga 模式設計的狀態機使用方式和原理。狀態機在我們的日常工作中使用非常廣泛，希望 Seata 的設計能對我們設計狀態機提供思路和參考。

大家好，我是君哥。

狀態機在我們的工作中應用非常廣泛，今天聊一聊分布式事務中間件 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：失敗狀態。

（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)。

（2）Choice

下面來看 ChoiceAccountState 這個狀態節點，Json 文件定義如下：

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

對應的下個節點是 ReduceAccount，如果失敗就會跳轉 Fail 狀態。

（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。

狀態機結束流程，提交或回滾全局事務。

6.高可用

Seata 中的狀態機并不是獨立部署，而是內嵌在應用中，由于狀態機上下文和執行日志都記錄在本地數據庫中，所以狀態機本身是無狀態的。

狀態機啟動時，會發送狀態到 Seata Server，當一個應用宕機后，Seata Server 能感知到，并會把恢復請求發送到存活的實例，收到請求的實例從數據庫取出狀態機上下文和執行日志進行恢復。如下圖：

7.總結

責任編輯：姜華來源：君哥聊技術

成人免费xxxxx在线视频软件_久久精品久久久_亚洲国产精品久久久_天天色天天色_亚洲人成一区_欧美一级欧美三级在线观看