GO编程DTM(二)
二阶段消息例子
本文将介绍一个完整的二阶段消息例子,让读者对二阶段消息型事务有一个准确的了解 业务场景
跨行转账是典型的分布式事务场景,在这里,A需要跨行转账给B,假设需求场景是:只有转出A可能失败,转入B是能够最终成功的 二阶段消息
二阶段消息是dtm首创的事务模式,用于替换本地事务表和事务消息这两种现有的方案。它能够保证本地事务的提交和全局事务提交是"原子的",适合解决不需要回滚的分布式事务场景。下面我们来看看二阶段消息,如何解决这个业务场景的问题。 核心业务
首先我们创建账户余额表: 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", `trading_balance` decimal(10,2) NOT NULL DEFAULT "0.00", `create_time` datetime DEFAULT now(), `update_time` datetime DEFAULT now() );
然后编写核心业务代码,调整用户的账户余额 func SagaAdjustBalance(db dtmcli.DB, uid int, amount int, result string) error { _, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?", amount, uid) return err }
再来编写具体的处理函数 app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { barrier := MustBarrierFromGin(c) return barrier.Call(txGet(), func(tx *sql.Tx) error { return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "") }) }))
这些处理函数的核心逻辑都是是调整余额。这里面的 barrier.Call 主要是用于处理幂等,保证重复调用不会多次调整余额,详情参见异常与子事务屏障二阶段消息事务
到此各个子事务的处理函数已经OK了,然后是开启二阶段消息事务,进行分支调用 msg := dtmcli.NewMsg(DtmServer, shortuuid.New()). Add(busi.Busi+"/SagaBTransIn", &TransReq{ Amount: 30 }) err := msg.DoAndSubmitDB(busi.Busi+"/QueryPreparedB", dbGet(), func(tx *sql.Tx) error { return busi.SagaAdjustBalance(tx, busi.TransOutUID, -req.Amount) })
这段代码中,会保证 DoAndSubmitDB 中的业务提交和全局事务提交是"原子的",保证了TransOut和TransIn的同时成功,或同时失败。其中 DoAndSubmitDB 中的第一个参数为回查URL,他的代码如下: app.GET(BusiAPI+"/QueryPreparedB", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.QueryPrepared(dbGet()) }))
至此,一个完整的二阶段消息分布式事务编写完成。 运行
如果您想要完整运行一个成功的示例,步骤如下: 运行dtm git clone https://github.com/dtm-labs/dtm && cd dtm go run main.go运行例子 git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples go run main.go http_msg_doAndCommit如何保证原子性
二阶段消息如何保证本地事务和全局事务要么都成功,要么都失败呢?假定本地事务提交完成后,提交全局事务前,进程crash会如何?下面时序图很好的讲解了二阶段消息是如何处理这个问题的:
图中的回查处理逻辑,dtm已经做了自动处理,用户只需要粘贴上述的代码即可 SAGA 例子
本文将介绍一个完整的 SAGA 例子,让读者对 SAGA 型事务有一个准确的了解 业务场景
跨行转账是典型的分布式事务场景,在这里,A需要跨行转账给B,假设需求场景是:转出A和转入B都有可能成功和失败,需要最终转入转出都成功,或者都失败 SAGA
Saga是这一篇数据库论文SAGAS提到的一个分布式事务方案。其核心思想是将长事务拆分为多个本地短事务,由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", `trading_balance` decimal(10,2) NOT NULL DEFAULT "0.00", `create_time` datetime DEFAULT now(), `update_time` datetime DEFAULT now() );
然后编写核心业务代码,调整用户的账户余额 func SagaAdjustBalance(db dtmcli.DB, uid int, amount int, result string) error { _, err := dtmimp.DBExec(db, "update dtm_busi.user_account set balance = balance + ? where user_id = ?", amount, uid) return err }
再来编写具体的正向操作/补偿操作的处理函数 app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { barrier := MustBarrierFromGin(c) return barrier.Call(txGet(), func(tx *sql.Tx) error { return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "") }) })) app.POST(BusiAPI+"/SagaBTransInCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { barrier := MustBarrierFromGin(c) return barrier.Call(txGet(), func(tx *sql.Tx) error { return SagaAdjustBalance(tx, TransInUID, -reqFrom(c).Amount, "") }) })) app.POST(BusiAPI+"/SagaBTransOut", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { barrier := MustBarrierFromGin(c) return barrier.Call(txGet(), func(tx *sql.Tx) error { return SagaAdjustBalance(tx, TransOutUID, -reqFrom(c).Amount, "") }) })) app.POST(BusiAPI+"/SagaBTransOutCom", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { barrier := MustBarrierFromGin(c) return barrier.Call(txGet(), func(tx *sql.Tx) error { return SagaAdjustBalance(tx, TransOutUID, reqFrom(c).Amount, "") }) }))
这些处理函数的核心逻辑都是是调整余额,对于这里面的 barrier.Call 作用,后面会详细解释SAGA 事务
到此各个子事务的处理函数已经OK了,然后是开启SAGA事务,进行分支调用 req := &gin.H{"amount": 30} // 微服务的载荷 // DtmServer为DTM服务的地址 saga := dtmcli.NewSaga(DtmServer, shortuuid.New()). // 添加一个TransOut的子事务,正向操作为url: qsBusi+"/TransOut", 逆向操作为url: qsBusi+"/TransOutCom" Add(qsBusi+"/SagaBTransOut", qsBusi+"/SagaBTransOutCom", req). // 添加一个TransIn的子事务,正向操作为url: qsBusi+"/TransOut", 逆向操作为url: qsBusi+"/TransInCom" Add(qsBusi+"/SagaBTransIn", qsBusi+"/SagaBTransInCom", req) // 提交saga事务,dtm会完成所有的子事务/回滚所有的子事务 err := saga.Submit()
至此,一个完整的SAGA分布式事务编写完成。 运行
如果您想要完整运行一个成功的示例,步骤如下: 运行dtm git clone https://github.com/dtm-labs/dtm && cd dtm go run main.go运行例子 git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples go run main.go http_saga_barrier
时序图如下:
处理网络异常
假设提交给dtm的事务中,调用转入操作时,出现短暂的故障怎么办?dtm 会重试未完成的操作,此时就会要求全局事务中的各个子事务是幂等的。dtm 框架首创子事务屏障技术,提供 BranchBarrier 工具类,可以帮助用户简单的处理幂等。它提供了一个函数 Call ,保证这个函数内部的业务,会被最多调用一次: func (bb *BranchBarrier) Call(tx *sql.Tx, busiCall BarrierBusiFunc) error
该 BranchBarrier 不仅能够自动处理幂等,还能够自动处理空补偿、悬挂的问题,详情可以参考异常与子事务屏障 处理回滚
假如银行将金额准备转入用户2时,发现用户2的账户异常,返回失败,会怎么样?我们调整处理函数,让转入操作返回失败 app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { return dtmcli.ErrFailure }))
我们给出事务失败交互的时序图
这里有一点,TransIn的正向操作什么都没有做,就返回了失败,此时调用TransIn的补偿操作,会不会导致反向调整出错了呢?
不用担心,前面的子事务屏障技术,能够保证TransIn的错误如果发生在提交之前,则补偿为空操作;TransIn的错误如果发生在提交之后,则补偿操作会将数据提交一次。
您可以将返回错误的TransIn改成: app.POST(BusiAPI+"/SagaBTransIn", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { barrier := MustBarrierFromGin(c) barrier.Call(txGet(), func(tx *sql.Tx) error { return SagaAdjustBalance(tx, TransInUID, reqFrom(c).Amount, "") }) return dtmcli.ErrFailure }))
最后的结果余额依旧会是对的,详情可以参考异常与子事务屏障 TCC 例子
本文将介绍一个完整的 TCC 例子,让读者对 TCC 型事务有一个准确的了解 业务场景
跨行转账是典型的分布式事务场景,在这里,A需要跨行转账给B,假设需求场景是:转出A和转入B都有可能成功和失败,需要最终转入转出都成功,或者都失败。
同时这里还有一个要求,假如发生回滚,SAGA 模式下会发生A发现自己的余额被扣减了,但是收款方B迟迟没有收到余额,那么会对A造成很大的困扰。业务上面希望不要出现这种情况 TCC组成
TCC分为3个阶段 Try 阶段:尝试执行,完成所有业务检查(一致性), 预留必须业务资源(准隔离性) Confirm 阶段:如果所有分支的Try都成功了,则走到Confirm阶段。Confirm真正执行业务,不作任何业务检查,只使用 Try 阶段预留的业务资源 Cancel 阶段:如果所有分支的Try有一个失败了,则走到Cancel阶段。Cancel释放 Try 阶段预留的业务资源。
如果我们要进行一个类似于银行跨行转账的业务,转出(TransOut)和转入(TransIn)分别在不同的微服务里,一个成功完成的TCC事务典型的时序图如下:
核心业务
首先我们创建账户余额表,其中 trading_balance 表示被冻结的金额: create table if not exists dtm_busi.user_account( id int(11) PRIMARY KEY AUTO_INCREMENT, user_id int(11) UNIQUE, balance DECIMAL(10, 2) not null default "0", trading_balance DECIMAL(10, 2) not null default "0", create_time datetime DEFAULT now(), update_time datetime DEFAULT now(), key(create_time), key(update_time) );
我们先编写核心代码,冻结/解冻资金操作,会检查约束balance+trading_balance >= 0,如果约束不成立,执行失败 func tccAdjustTrading(db dtmcli.DB, uid int, amount int) error { affected, err := dtmimp.DBExec(db, `update dtm_busi.user_account set trading_balance=trading_balance+? where user_id=? and trading_balance + ? + balance >= 0`, amount, uid, amount) if err == nil && affected == 0 { return fmt.Errorf("update error, maybe balance not enough") } return err } func tccAdjustBalance(db dtmcli.DB, uid int, amount int) error { affected, err := dtmimp.DBExec(db, `update dtm_busi.user_account set trading_balance=trading_balance-?, balance=balance+? where user_id=?`, amount, amount, uid) if err == nil && affected == 0 { return fmt.Errorf("update user_account 0 rows") } return err }
下面我们来编写具体的Try/Confirm/Cancel的处理函数 app.POST(BusiAPI+"/TccBTransOutTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransOutUID, -req.Amount) }) })) app.POST(BusiAPI+"/TccBTransOutConfirm", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustBalance(tx, TransOutUID, -reqFrom(c).Amount) }) })) app.POST(BusiAPI+"/TccBTransOutCancel", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransOutUID, req.Amount) }) })) app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransInUID, req.Amount) }) })) app.POST(BusiAPI+"/TccBTransOutConfirm", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustBalance(tx, TransInUID, reqFrom(c).Amount) }) })) app.POST(BusiAPI+"/TccBTransInCancel", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) return bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransInUID, -req.Amount) }) }))
到此各个子事务的处理函数已经OK了,这些处理函数的核心逻辑都是冻结和调整余额,对于这里面的 bb.Call 作用,后面会详细解释TCC 事务
然后是开启TCC事务,进行分支调用 // TccGlobalTransaction 会开启一个全局事务 _, err := dtmcli.TccGlobalTransaction(DtmServer, func(tcc *dtmcli.Tcc) (rerr error) { // CallBranch 会将事务分支的Confirm/Cancel注册到全局事务上,然后直接调用Try res1, rerr := tcc.CallBranch(&TransReq{Amount: 30}, host+"/api/TccBTransOutTry", host+"/api/TccBTransOutConfirm", host+"/api/TccBTransOutCancel" if err != nil { return resp, err } return tcc.CallBranch(&TransReq{Amount: 30}, host+"/api/TccBTransInTry", host+"/api/TccBTransInConfirm", host+"/api/TccBTransInCancel") })
至此,一个完整的TCC分布式事务编写完成。 运行
如果您想要完整运行一个成功的示例,步骤如下: 运行dtm git clone https://github.com/dtm-labs/dtm && cd dtm go run main.go运行例子 git clone https://github.com/dtm-labs/dtm-examples && cd dtm-examples go run main.go http_tcc_barrier处理网络异常
假设提交给dtm的事务中,这些步骤中,出现短暂的故障怎么办?dtm 会重试未完成的操作,此时就会要求全局事务中的各个子事务是幂等的。dtm 框架首创子事务屏障技术,提供 BranchBarrier 工具类,可以帮助用户简单的处理幂等。它提供了一个函数 Call ,保证这个函数内部的业务,会被最多调用一次: func (bb *BranchBarrier) Call(tx *sql.Tx, busiCall BarrierBusiFunc) error
该 BranchBarrier 不仅能够自动处理幂等,还能够自动处理空补偿、悬挂的问题,详情可以参考异常与子事务屏障 TCC的回滚
假如银行将金额准备转入用户2时,发现用户2的账户异常,返回失败,会怎么样?我们修改代码,模拟这种情况: app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { return dtmcli.ErrFailure }))
这是事务失败交互的时序图
这个跟成功的TCC差别就在于,当某个子事务返回失败后,后续就回滚全局事务,调用各个子事务的Cancel操作,保证全局事务全部回滚。
这里有一点,TransInTry的正向操作什么都没有做,就返回了失败,此时调用TransInCancel补偿操作,会不会导致反向调整出错了呢?
不用担心,前面的子事务屏障技术,能够保证TransInTry的错误如果发生在提交之前,则补偿为空操作;TransInTry的错误如果发生在提交之后,则补偿操作会将数据提交一次。
您可以将 TccBTransInTry 改成app.POST(BusiAPI+"/TccBTransInTry", dtmutil.WrapHandler2(func(c *gin.Context) interface{} { bb := MustBarrierFromGin(c) bb.Call(txGet(), func(tx *sql.Tx) error { return tccAdjustTrading(tx, TransInUID, req.Amount) }) return dtmcli.ErrFailure }))
最后的结果余额依旧会是对的,详情可以参考异常与子事务屏障
江一平为冈村宁次作无罪辩护的中国律师,晚年家破人亡无人照顾1937年的秋天,中国迎来了一群不速之客。他们在中国的土地上烧杀抢掠无恶不作。无数的同胞命丧于此,无数的妇女儿童惨遭凌辱。而他们有一个共同的名字,那就是日本鬼子。虽然经过我国先辈们
塞尔维亚与科索沃剑拔弩张,塞尔维亚被逼角落历史上引发欧洲多场战争的火药桶,再次有爆炸的危险。这个火药桶,就是科索沃局势。中国的80后和90后这一代人,应该多少都保留着科索沃危机的记忆。科索沃危机于1998年正式爆发,北约在
侵华日军的两封家书众所周知,日本侵略者曾经在中国烧杀抢掠,无恶不作。事实上,在战败后,日本也有些人拒不道歉,甚至否认曾经犯下的罪行。面对血腥的历史和扎实的证据,他们紧闭双眼,口中念着这是一面之词,心
康熙第五子胤祺,宠妃生太后养,不争不抢,得雍正善待!了解清史的人都知道康熙中后期,尤其是太子胤礽被废后,诸皇子明争暗斗争夺储君之位,一般我们都称之为九王夺嫡之所以称之为九王夺嫡,就是因为一共有九位皇子参与了争储的斗争皇长子胤褆皇次子
侯景之乱重创南朝梁,加剧了南弱北强的形势!头条创作挑战赛侯景之乱,又称太清之难,是指南北朝时期南朝梁将领侯景发动的叛乱事件。侯景本为东魏叛将,被梁武帝萧衍所收留,因对梁朝与东魏通好心怀不满,遂于548年以清君侧为名义在寿阳
荣氏做人钟商标,是为警示国人吗?(1)人钟,是荣氏家族企业的主要品牌,解放前家喻户晓。其图案,是一个孩童拿着钟绳在敲一只硕大的钟。以人钟做产品商标,其寓意何在?学界研究荣氏的文字,提到人钟,多以警醒国人做解读。笔者以为
如皋王惟熙(11)王惟熙三任贡举考官的那些进士十一王惟熙三任贡举考官的那些进士嘉祐六年(1061)贡举的授职条例是这样的宋会要辑稿选举二记载嘉祐六年四月二十二日,以新及第进士第一人王俊民为大理评事签书雄武军节度判官厅公事,第二
李鸿章是一个怎样的人李鸿章是一个怎样的人呢?有人说他有创新精神,有远大的目标。但是他这个人软弱无能,贪图享受。疯狂的贪污敛财完全是他个人自私自利的表现。为什么各国侵略者都指名道姓要李鸿章去谈判呢?就是
空难,兴登堡号飞艇坠落1872年,法国人特罗姆成功制成了一艘螺旋桨飞艇,这是一种新的飞行器,它不同于飞机,它是利用轻于空气的气体来提供升力的航空器,又有一个充满氢气的气囊作浮力,有点类似气球。但与气球相
中美粮食战争,美国四大粮商想做空中国,却败于中储粮之手谁控制了石油,就控制了所有国家谁控制了粮食,就控制了人类谁控制了货币,就控制了全球经济。前美国国务卿国际战略问题专家亨利艾尔弗雷德基辛格。粮食石油美元是美国称霸全球的三大战略,而事
第二次洋务运动一般意义上的洋务运动是指18601895年的第一次洋务运动。六王爷爱新觉罗奕訢上奏的通筹夷务全局酌拟章程六条,标志着洋务运动的开始。实干的洋务派代表是曾国藩李鸿章左宗棠张之洞刘坤一