带你用Python实现Saga 分布式事务的方法
作者:叶东富 发布时间:2022-02-17 06:17:11
银行跨行转账业务是一个典型分布式事务场景,假设 A 需要跨行转账给 B,那么就涉及两个银行的数据,无法通过一个数据库的本地事务保证转账的 ACID,只能够通过分布式事务来解决。
分布式事务
分布式事务在分布式环境下,为了满足可用性、性能与降级服务的需要,降低一致性与隔离性的要求,一方面遵循 BASE 理论:
基本业务可用性( Basic Availability )
柔性状态( Soft state )
最终一致性( Eventual consistency )
另一方面,分布式事务也部分遵循 ACID 规范:
原子性:严格遵循
一致性:事务完成后的一致性严格遵循;事务中的一致性可适当放宽
隔离性:并行事务间不可影响;事务中间结果可见性允许安全放宽
持久性:严格遵循
SAGA
Saga 是这一篇数据库论文SAGAS提到的一个分布式事务方案。其核心思想是将长事务拆分为多个本地短事务,由 Saga 事务协调器协调,如果各个本地事务成功完成那就正常完成,如果某个步骤失败,则根据相反顺序一次调用补偿操作。
目前可用于 SAGA 的开源框架,主要为 Java 语言,其中以 seata 为代表。我们的例子采用 go 语言,使用的分布式事务框架为https://github.com/yedf/dtm,它对分布式事务的支持非常优雅。下面来详细讲解 SAGA 的组成:
DTM 事务框架里,有 3 个角色,与经典的 XA 分布式事务一样:
AP/应用程序,发起全局事务,定义全局事务包含哪些事务分支
RM/资源管理器,负责分支事务各项资源的管理
TM/事务管理器,负责协调全局事务的正确执行,包括 SAGA 正向 /逆向操作的执行
下面看一个成功完成的 SAGA 时序图,就很容易理解 SAGA 分布式事务:
SAGA实践
对于我们要进行的银行转账的例子,我们将在正向操作中,进行转入转出,在补偿操作中,做相反的调整。
首先我们创建账户余额表:
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()
);
我们先编写核心业务代码,调整用户的账户余额
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")
下面我们来编写具体的正向操作 /补偿操作的处理函数
@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"}
到此各个子事务的处理函数已经 OK 了,然后是开启 SAGA 事务,进行分支调用
# 这是 dtm 服务地址
dtm = "http://localhost:8080/api/dtmsvr"
# 这是业务微服务地址
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()
至此,一个完整的 SAGA 分布式事务编写完成。
如果您想要完整运行一个成功的示例,那么参考这个例子yedf/dtmcli-py-sample,将它运行起来非常简单
# 部署启动 dtm
# 需要 docker 版本 18 以上
git clone https://github.com/yedf/dtm
cd dtm
docker-compose up
# 另起一个命令行
git clone https://github.com/yedf/dtmcli-py-sample
cd dtmcli-py-sample
pip3 install flask dtmcli requests
flask run
# 另起一个命令行
curl localhost:5000/api/fireSaga
处理网络异常
假设提交给 dtm 的事务中,调用转入操作时,出现短暂的故障怎么办?按照 SAGA 事务的协议,dtm 会重试未完成的操作,这时我们要如何处理?故障有可能是转入操作完成后出网络故障,也有可能是转入操作完成中出现机器宕机。如何处理才能够保障账户余额的调整是正确无问题的?
这类网络异常的妥当处理,是分布式事务中的大难题,异常情况包括三类:重复请求、空补偿、悬挂,都需要正确处理
DTM 提供了子事务屏障功能,保证上述异常情况下的业务逻辑,只会有一次正确顺序下的成功提交。(子事务屏障详情参考分布式事务最经典的七种解决方案的子事务屏障环节)
我们把处理函数调整为:
@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)调用会使用子事务屏障技术,保证 busi_callback 回调函数仅被提交一次
您可以尝试多次调用这个 TransIn 服务,仅有一次余额调整。
处理回滚
假如银行将金额准备转入用户 2 时,发现用户 2 的账户异常,返回失败,会怎么样?我们调整处理函数,让转入操作返回失败
@app.post("/api/TransInSaga")
def trans_in_saga():
return {"dtm_result": "FAILURE"}
我们给出事务失败交互的时序图
这里有一点,TransIn 的正向操作什么都没有做,就返回了失败,此时调用 TransIn 的补偿操作,会不会导致反向调整出错了呢?
不用担心,前面的子事务屏障技术,能够保证 TransIn 的错误如果发生在提交之前,则补偿为空操作;TransIn 的错误如果发生在提交之后,则补偿操作会将数据提交一次。
您可以将返回错误的 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"}
最后的结果余额依旧会是对的,原理可以参考:分布式事务最经典的七种解决方案的子事务屏障环节
小结
在这篇文章里,我们介绍了 SAGA 的理论知识,也通过一个例子,完整给出了编写一个 SAGA 事务的过程,涵盖了正常成功完成,异常情况,以及成功回滚的情况。相信读者通过这边文章,对 SAGA 已经有了深入的理解。
文中使用的 dtm 是新开源的 Golang 分布式事务管理框架,功能强大,支持 TCC 、SAGA 、XA 、事务消息等事务模式,支持 Go 、python 、PHP 、node 、csharp 等语言的。同时提供了非常简单易用的接口。
阅读完此篇干货,欢迎大家访问项目https://github.com/yedf/dtm,给颗星星支持!
来源:https://developer.51cto.com/art/202109/681073.htm
猜你喜欢
- 最近稍稍有点空闲时间,于是重新温习了一下之前学习过的python基础。废话不多说,记录一下自己的所得。首先,安装什么的不在本人的温习范围,另
- 本篇内容介绍了Python编程语言的优势和不足。首先我们来了解什么是Python?Python是一种面向对象的高级编程语言,具有集成的动态语
- 1、数据驱动介绍:@ddt.ddt(类装饰器,申明当前类使用ddt框架)@ddt.data(函数装饰器,用于给测试用例传递数据),支持传py
- 通过HTTP_USER_AGENT判断用户是从手机上访问,还是电脑IE上访问。 asp代码片段:主要使用了正则匹配手机环境,大家可以补充手机
- session_unset() 释放当前在内存中已经创建的所有$_SESSION变量,但不删除session文件以及不释放对应的sessio
- 许多服务器管理员都知道,MySQL数据库管理系统(RDBMS)是高度灵活的软件块,带有范围广阔的启动选项,可以用来修改相关行为。然而,大部分
- 在注册或购物车结帐的时候,需要用户填入个人资料,这个环节可以简练一下,我们只需客人填入邮政编码,然后就根据这个邮政编码,自动从数据库中取出相
- 一般情况下只有需要长期运行的项目才会去关注内存的增长情况,即使是很小部分的内存泄露经过长期的运行仍然会产生很大的隐患。python本身也是支
- django 返回数据的主要是用django.http.HttpResponse 中的HttpResponse 完成的具体的我直接贴代码吧
- 工作中,网页设计师经常会遇见这些状况:时间这么短又要出彩、又是要大气要有气氛、风格不明确很难把握、栏目这么多页面又这么长……突然觉得束手无策
- Django项目中为什么会加载静态时会失败呢?原因:django部署方式比较特别,采用静态文件路径:STATICFILES_DIRS的部署方
- 我们知道 Pandas 是数据科学社区中流行的 Python 包,它包含许多函数和方法来分析数据。尽管它的功能对于数据分析来说足够有效,但定
- 方法一先按esc进入命令模式,即左侧线为蓝色(为绿色时是编辑模式),按z键即可恢复方法二如果是运行过的代码直接运行history方法三功能栏
- PHP get_html_translation_table() 函数实例输出 htmlspecialchars 函数使用的翻译表:<
- 本文实例为大家分享了答题辅助python具体代码,供大家参考,具体内容如下from screenshot import pull_scree
- https://docs.python.org/3/library/function.html #python官方网址# 取绝对值print
- 需求:主线程开启了多个线程去干活,每个线程需要完成的时间不同,但是在干完活以后都要通知给主线程下面上代码:#!/usr/bin/python
- 项目介绍背景:DC竞赛比赛项目,运用回归模型进 * 价预测。数据介绍:数据主要包括2014年5月至2015年5月美国King County的房
- 案例:该数据集的是一个关于每个学生成绩的数据集,接下来我们对该数据集进行分析,判断学生是否适合继续深造数据集特征展示1 GRE
- 本文以实例形式讲述了Python中切片操作的用法,分享给大家供大家参考借鉴,具体如下:取一个list或tuple的部分元素是非常常见的操作。