講解如何利用 Python完成 Saga 分布式事務(wù)

銀行跨行轉(zhuǎn)賬業(yè)務(wù)是一個(gè)典型分布式事務(wù)場景，假設(shè) A 需要跨行轉(zhuǎn)賬給 B，那么就涉及兩個(gè)銀行的數(shù)據(jù)，無法通過一個(gè)數(shù)據(jù)庫的本地事務(wù)保證轉(zhuǎn)賬的 ACID，只能夠通過分布式事務(wù)來解決。

1、分布式事務(wù)

分布式事務(wù)在分布式環(huán)境下，為了滿足可用性、性能與降級(jí)服務(wù)的需要，降低一致性與隔離性的要求，一方面遵循 BASE 理論：

• 基本業(yè)務(wù)可用性（ Basic Availability ）
• 柔性狀態(tài)（ Soft state ）
• 最終一致性（ Eventual consistency ）

另一方面，分布式事務(wù)也部分遵循 ACID 規(guī)范：

• 原子性：嚴(yán)格遵循
• 一致性：事務(wù)完成后的一致性嚴(yán)格遵循；事務(wù)中的一致性可適當(dāng)放寬
• 隔離性：并行事務(wù)間不可影響；事務(wù)中間結(jié)果可見性允許安全放寬
• 持久性：嚴(yán)格遵循

2、SAGA

Saga 是這一篇數(shù)據(jù)庫論文SAGAS提到的一個(gè)分布式事務(wù)方案。其核心思想是將長事務(wù)拆分為多個(gè)本地短事務(wù)，由 Saga 事務(wù)協(xié)調(diào)器協(xié)調(diào)，如果各個(gè)本地事務(wù)成功完成那就正常完成，如果某個(gè)步驟失敗，則根據(jù)相反順序一次調(diào)用補(bǔ)償操作。

目前可用于 SAGA 的開源框架，主要為 Java 語言，其中以 seata 為代表。我們的例子采用 go 語言，使用的分布式事務(wù)框架為https://github.com/yedf/dtm，它對(duì)分布式事務(wù)的支持非常優(yōu)雅。下面來詳細(xì)講解 SAGA 的組成：

DTM 事務(wù)框架里，有 3 個(gè)角色，與經(jīng)典的 XA 分布式事務(wù)一樣：

• AP/應(yīng)用程序，發(fā)起全局事務(wù)，定義全局事務(wù)包含哪些事務(wù)分支
• RM/資源管理器，負(fù)責(zé)分支事務(wù)各項(xiàng)資源的管理
• TM/事務(wù)管理器，負(fù)責(zé)協(xié)調(diào)全局事務(wù)的正確執(zhí)行，包括 SAGA 正向 /逆向操作的執(zhí)行

下面看一個(gè)成功完成的 SAGA 時(shí)序圖，就很容易理解 SAGA 分布式事務(wù)：

3、SAGA 實(shí)踐

對(duì)于我們要進(jìn)行的銀行轉(zhuǎn)賬的例子，我們將在正向操作中，進(jìn)行轉(zhuǎn)入轉(zhuǎn)出，在補(bǔ)償操作中，做相反的調(diào)整。

首先我們創(chuàng)建賬戶余額表：

CREATE TABLE dtm_busi.`user_account` ( 
  `id` int(11) AUTO_INCREMENT PRIMARY KEY, 
  `user_id` int(11) not NULL UNIQUE , 
  `balance` decimal(10,2) NOT NULL DEFAULT '0.00', 
  `create_time` datetime DEFAULT now(), 
  `update_time` datetime DEFAULT now() 
); 

我們先編寫核心業(yè)務(wù)代碼，調(diào)整用戶的賬戶余額

def saga_adjust_balance(cursor, uid, amount): 
  affected = utils.sqlexec(cursor, "update dtm_busi.user_account set balance=balance+%d where user_id=%d and balance >= -%d" %(amount, uid, amount)) 
  if affected == 0: 
    raise Exception("update error, balance not enough") 

下面我們來編寫具體的正向操作 /補(bǔ)償操作的處理函數(shù)

@app.post("/api/TransOutSaga") 
def trans_out_saga(): 
  saga_adjust_balance(c, out_uid, -30) 
  return {"dtm_result": "SUCCESS"} 
 
@app.post("/api/TransOutCompensate") 
def trans_out_compensate(): 
  saga_adjust_balance(c, out_uid, 30) 
  return {"dtm_result": "SUCCESS"} 
 
@app.post("/api/TransInSaga") 
def trans_in_saga(): 
  saga_adjust_balance(c, in_uid, 30) 
  return {"dtm_result": "SUCCESS"} 
 
@app.post("/api/TransInCompensate") 
def trans_in_compensate(): 
  saga_adjust_balance(c, in_uid, -30) 
  return {"dtm_result": "SUCCESS"} 

到此各個(gè)子事務(wù)的處理函數(shù)已經(jīng) OK 了，然后是開啟 SAGA 事務(wù)，進(jìn)行分支調(diào)用

# 這是 dtm 服務(wù)地址 
dtm = "http://localhost:8080/api/dtmsvr" 
# 這是業(yè)務(wù)微服務(wù)地址 
svc = "http://localhost:5000/api" 
 
    req = {"amount": 30} 
    s = saga.Saga(dtm, utils.gen_gid(dtm)) 
    s.add(req, svc + "/TransOutSaga", svc + "/TransOutCompensate") 
    s.add(req, svc + "/TransInSaga", svc + "/TransInCompensate") 
    s.submit() 

至此，一個(gè)完整的 SAGA 分布式事務(wù)編寫完成。

如果您想要完整運(yùn)行一個(gè)成功的示例，那么參考這個(gè)例子yedf/dtmcli-py-sample，將它運(yùn)行起來非常簡單

# 部署啟動(dòng) dtm 
# 需要 docker 版本 18 以上 
git clone https://github.com/yedf/dtm 
cd dtm 
docker-compose up 
 
# 另起一個(gè)命令行 
git clone https://github.com/yedf/dtmcli-py-sample 
cd dtmcli-py-sample 
pip3 install flask dtmcli requests 
flask run 
 
# 另起一個(gè)命令行 
curl localhost:5000/api/fireSaga 

4、處理網(wǎng)絡(luò)異常

假設(shè)提交給 dtm 的事務(wù)中，調(diào)用轉(zhuǎn)入操作時(shí)，出現(xiàn)短暫的故障怎么辦？按照 SAGA 事務(wù)的協(xié)議，dtm 會(huì)重試未完成的操作，這時(shí)我們要如何處理？故障有可能是轉(zhuǎn)入操作完成后出網(wǎng)絡(luò)故障，也有可能是轉(zhuǎn)入操作完成中出現(xiàn)機(jī)器宕機(jī)。如何處理才能夠保障賬戶余額的調(diào)整是正確無問題的？

這類網(wǎng)絡(luò)異常的妥當(dāng)處理，是分布式事務(wù)中的大難題，異常情況包括三類：重復(fù)請(qǐng)求、空補(bǔ)償、懸掛，都需要正確處理

DTM 提供了子事務(wù)屏障功能，保證上述異常情況下的業(yè)務(wù)邏輯，只會(huì)有一次正確順序下的成功提交。（子事務(wù)屏障詳情參考分布式事務(wù)最經(jīng)典的七種解決方案的子事務(wù)屏障環(huán)節(jié)）

我們把處理函數(shù)調(diào)整為：

@app.post("/api/TransOutSaga") 
def trans_out_saga(): 
  with barrier.AutoCursor(conn_new()) as cursor: 
    def busi_callback(c): 
      saga_adjust_balance(c, out_uid, -30) 
    barrier_from_req(request).call(cursor, busi_callback) 
  return {"dtm_result": "SUCCESS"} 

這里的 barrier_from_req(request).call(cursor, busi_callback)調(diào)用會(huì)使用子事務(wù)屏障技術(shù)，保證 busi_callback 回調(diào)函數(shù)僅被提交一次

您可以嘗試多次調(diào)用這個(gè) TransIn 服務(wù)，僅有一次余額調(diào)整。

5、處理回滾

假如銀行將金額準(zhǔn)備轉(zhuǎn)入用戶 2 時(shí)，發(fā)現(xiàn)用戶 2 的賬戶異常，返回失敗，會(huì)怎么樣？我們調(diào)整處理函數(shù)，讓轉(zhuǎn)入操作返回失敗

@app.post("/api/TransInSaga") 
def trans_in_saga(): 
  return {"dtm_result": "FAILURE"} 

我們給出事務(wù)失敗交互的時(shí)序圖：

這里有一點(diǎn)，TransIn 的正向操作什么都沒有做，就返回了失敗，此時(shí)調(diào)用 TransIn 的補(bǔ)償操作，會(huì)不會(huì)導(dǎo)致反向調(diào)整出錯(cuò)了呢？

不用擔(dān)心，前面的子事務(wù)屏障技術(shù)，能夠保證 TransIn 的錯(cuò)誤如果發(fā)生在提交之前，則補(bǔ)償為空操作；TransIn 的錯(cuò)誤如果發(fā)生在提交之后，則補(bǔ)償操作會(huì)將數(shù)據(jù)提交一次。

我們可以將返回錯(cuò)誤的 TransIn 改成：

@app.post("/api/TransInSaga") 
def trans_in_saga(): 
  with barrier.AutoCursor(conn_new()) as cursor: 
    def busi_callback(c): 
      saga_adjust_balance(c, in_uid, 30) 
    barrier_from_req(request).call(cursor, busi_callback) 
  return {"dtm_result": "FAILURE"} 

最后的結(jié)果余額依舊會(huì)是對(duì)的，原理可以參考：分布式事務(wù)最經(jīng)典的七種解決方案的子事務(wù)屏障環(huán)節(jié)

6、小結(jié)

在這篇文章里，我們介紹了 SAGA 的理論知識(shí)，也通過一個(gè)例子，完整給出了編寫一個(gè) SAGA 事務(wù)的過程，涵蓋了正常成功完成，異常情況，以及成功回滾的情況。相信讀者通過這邊文章，對(duì) SAGA 已經(jīng)有了深入的理解。

文中使用的 dtm 是新開源的 Golang 分布式事務(wù)管理框架，功能強(qiáng)大，支持 TCC 、SAGA 、XA 、事務(wù)消息等事務(wù)模式，支持 Go 、python 、PHP 、node 、csharp 等語言的。同時(shí)提供了非常簡單易用的接口。

以上就是利用 Python 輕松完成一個(gè) Saga 分布式事務(wù)的詳細(xì)內(nèi)容，更多關(guān)于Python完成一個(gè) Saga 分布式事務(wù)的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！