mirror of https://github.com/dtm-labs/dtm.git
yedongfu
5 years ago
17 changed files with 588 additions and 309 deletions
@ -0,0 +1,69 @@ |
|||
package dtm |
|||
|
|||
import ( |
|||
"context" |
|||
"database/sql" |
|||
"fmt" |
|||
|
|||
"github.com/yedf/dtm/common" |
|||
) |
|||
|
|||
type BusiFunc func(db *sql.DB) (interface{}, error) |
|||
|
|||
type TransInfo struct { |
|||
TransType string |
|||
Gid string |
|||
BranchID string |
|||
BranchType string |
|||
} |
|||
|
|||
func (t *TransInfo) String() string { |
|||
return fmt.Sprintf("transInfo: %s %s %s %s", t.TransType, t.Gid, t.BranchID, t.BranchType) |
|||
} |
|||
|
|||
type BarrierModel struct { |
|||
common.ModelBase |
|||
TransInfo |
|||
} |
|||
|
|||
func (BarrierModel) TableName() string { return "dtm_barrier.barrier" } |
|||
|
|||
func insertBarrier(tx *sql.Tx, transType string, gid string, branchID string, branchType string) (int64, error) { |
|||
if branchType == "" { |
|||
return 0, nil |
|||
} |
|||
res, err := tx.Exec("insert into dtm_barrier.barrier(trans_type, gid, branch_id, branch_type) values(?,?,?,?)", transType, gid, branchID, branchType) |
|||
if err != nil { |
|||
return 0, err |
|||
} |
|||
return res.RowsAffected() |
|||
} |
|||
|
|||
func ThroughBarrierCall(db *sql.DB, transType string, gid string, branchId string, branchType string, busiCall BusiFunc) (res interface{}, rerr error) { |
|||
tx, rerr := db.BeginTx(context.Background(), &sql.TxOptions{}) |
|||
if rerr != nil { |
|||
return |
|||
} |
|||
defer func() { |
|||
if x := recover(); x != nil { |
|||
tx.Rollback() |
|||
panic(x) |
|||
} else if rerr != nil { |
|||
tx.Rollback() |
|||
} else { |
|||
tx.Commit() |
|||
} |
|||
}() |
|||
|
|||
originType := map[string]string{ |
|||
"cancel": "action", |
|||
"compensate": "action", |
|||
}[branchType] |
|||
originAffected, _ := insertBarrier(tx, transType, gid, branchId, originType) |
|||
currentAffected, rerr := insertBarrier(tx, transType, gid, branchId, branchType) |
|||
if currentAffected == 0 || (originType == "cancel" || originType == "compensate") && originAffected > 0 { |
|||
return |
|||
} |
|||
res, rerr = busiCall(db) |
|||
return |
|||
} |
|||
@ -0,0 +1,88 @@ |
|||
package examples |
|||
|
|||
import ( |
|||
"fmt" |
|||
"time" |
|||
|
|||
"github.com/gin-gonic/gin" |
|||
"github.com/sirupsen/logrus" |
|||
"github.com/yedf/dtm/common" |
|||
) |
|||
|
|||
const ( |
|||
BusiApi = "/api/busi" |
|||
BusiPort = 8081 |
|||
) |
|||
|
|||
var Busi string = fmt.Sprintf("http://localhost:%d%s", BusiPort, BusiApi) |
|||
|
|||
func BaseAppNew() *gin.Engine { |
|||
logrus.Printf("examples starting") |
|||
app := common.GetGinApp() |
|||
return app |
|||
} |
|||
|
|||
func BaseAppStart(app *gin.Engine) { |
|||
logrus.Printf("Starting busi at: %d", BusiPort) |
|||
go app.Run(fmt.Sprintf(":%d", BusiPort)) |
|||
time.Sleep(100 * time.Millisecond) |
|||
} |
|||
|
|||
type AutoEmptyString struct { |
|||
value string |
|||
} |
|||
|
|||
func (s *AutoEmptyString) SetOnce(v string) { |
|||
s.value = v |
|||
} |
|||
|
|||
func (s *AutoEmptyString) Fetch() string { |
|||
v := s.value |
|||
s.value = "" |
|||
return v |
|||
} |
|||
|
|||
type mainSwitchType struct { |
|||
TransInResult AutoEmptyString |
|||
TransOutResult AutoEmptyString |
|||
TransInConfirmResult AutoEmptyString |
|||
TransOutConfirmResult AutoEmptyString |
|||
TransInRevertResult AutoEmptyString |
|||
TransOutRevertResult AutoEmptyString |
|||
CanSubmitResult AutoEmptyString |
|||
} |
|||
|
|||
var MainSwitch mainSwitchType |
|||
|
|||
func handleGeneralBusiness(c *gin.Context, result1 string, result2 string, busi string) (interface{}, error) { |
|||
info := infoFromContext(c) |
|||
res := common.OrString(MainSwitch.TransInResult.Fetch(), result2, "SUCCESS") |
|||
logrus.Printf("%s %s result: %s", info.String(), common.GetFuncName(), res) |
|||
return M{"result": res}, nil |
|||
|
|||
} |
|||
|
|||
func BaseAppSetup(app *gin.Engine) { |
|||
app.POST(BusiApi+"/TransIn", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
return handleGeneralBusiness(c, MainSwitch.TransInResult.Fetch(), reqFrom(c).TransInResult, "transIn") |
|||
})) |
|||
app.POST(BusiApi+"/TransOut", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
return handleGeneralBusiness(c, MainSwitch.TransOutResult.Fetch(), reqFrom(c).TransOutResult, "transIn") |
|||
})) |
|||
app.POST(BusiApi+"/TransInConfirm", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
return handleGeneralBusiness(c, MainSwitch.TransInConfirmResult.Fetch(), "", "transIn") |
|||
})) |
|||
app.POST(BusiApi+"/TransOutConfirm", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
return handleGeneralBusiness(c, MainSwitch.TransOutConfirmResult.Fetch(), "", "transIn") |
|||
})) |
|||
app.POST(BusiApi+"/TransInRevert", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
return handleGeneralBusiness(c, MainSwitch.TransInRevertResult.Fetch(), "", "transIn") |
|||
})) |
|||
app.POST(BusiApi+"/TransOutRevert", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
return handleGeneralBusiness(c, MainSwitch.TransOutRevertResult.Fetch(), "", "transIn") |
|||
})) |
|||
app.GET(BusiApi+"/CanSubmit", common.WrapHandler(func(c *gin.Context) (interface{}, error) { |
|||
logrus.Printf("%s CanSubmit", c.Query("gid")) |
|||
return common.OrString(MainSwitch.CanSubmitResult.Fetch(), "SUCCESS"), nil |
|||
})) |
|||
} |
|||
@ -0,0 +1,101 @@ |
|||
package examples |
|||
|
|||
import ( |
|||
"database/sql" |
|||
"fmt" |
|||
"time" |
|||
|
|||
"github.com/gin-gonic/gin" |
|||
"github.com/sirupsen/logrus" |
|||
"github.com/yedf/dtm" |
|||
"github.com/yedf/dtm/common" |
|||
"gorm.io/gorm" |
|||
) |
|||
|
|||
// 事务参与者的服务地址
|
|||
const SagaBarrierBusiApi = "/api/busi_saga_barrier" |
|||
|
|||
var SagaBarrierBusi = fmt.Sprintf("http://localhost:%d%s", SagaBarrierBusiPort, SagaBarrierBusiApi) |
|||
|
|||
func SagaBarrierMain() { |
|||
go SagaBarrierStartSvr() |
|||
SagaBarrierFireRequest() |
|||
time.Sleep(1000 * time.Second) |
|||
} |
|||
|
|||
func SagaBarrierStartSvr() { |
|||
logrus.Printf("saga barrier examples starting") |
|||
app := common.GetGinApp() |
|||
SagaBarrierAddRoute(app) |
|||
app.Run(fmt.Sprintf(":%d", SagaBarrierBusiPort)) |
|||
} |
|||
|
|||
func SagaBarrierFireRequest() { |
|||
logrus.Printf("a busi transaction begin") |
|||
req := &TransReq{ |
|||
Amount: 30, |
|||
TransInResult: "SUCCESS", |
|||
TransOutResult: "SUCCESS", |
|||
} |
|||
saga := dtm.SagaNew(DtmServer). |
|||
Add(SagaBarrierBusi+"/TransOut", SagaBarrierBusi+"/TransOutCompensate", req). |
|||
Add(SagaBarrierBusi+"/TransIn", SagaBarrierBusi+"/TransInCompensate", req) |
|||
logrus.Printf("busi trans commit") |
|||
err := saga.Commit() |
|||
e2p(err) |
|||
} |
|||
|
|||
// api
|
|||
|
|||
func SagaBarrierAddRoute(app *gin.Engine) { |
|||
app.POST(SagaBarrierBusiApi+"/TransIn", common.WrapHandler(sagaBarrierTransIn)) |
|||
app.POST(SagaBarrierBusiApi+"/TransInCompensate", common.WrapHandler(sagaBarrierTransInCompensate)) |
|||
app.POST(SagaBarrierBusiApi+"/TransOut", common.WrapHandler(sagaBarrierTransOut)) |
|||
app.POST(SagaBarrierBusiApi+"/TransOutCompensate", common.WrapHandler(sagaBarrierTransOutCompensate)) |
|||
logrus.Printf("examples listening at %d", SagaBarrierBusiPort) |
|||
} |
|||
|
|||
var SagaBarrierTransInResult = "" |
|||
var SagaBarrierTransOutResult = "" |
|||
var SagaBarrierTransInCompensateResult = "" |
|||
var SagaBarrierTransOutCompensateResult = "" |
|||
|
|||
func sagaBarrierTransIn(c *gin.Context) (interface{}, error) { |
|||
gid := c.Query("gid") |
|||
req := reqFrom(c) |
|||
res := common.OrString(SagaBarrierTransInResult, req.TransInResult, "SUCCESS") |
|||
logrus.Printf("%s TransIn: %v result: %s", gid, req, res) |
|||
return M{"result": res}, nil |
|||
} |
|||
|
|||
func sagaBarrierTransInCompensate(c *gin.Context) (interface{}, error) { |
|||
gid := c.Query("gid") |
|||
req := reqFrom(c) |
|||
res := common.OrString(SagaBarrierTransInCompensateResult, "SUCCESS") |
|||
logrus.Printf("%s TransInCompensate: %v result: %s", gid, req, res) |
|||
return M{"result": res}, nil |
|||
} |
|||
|
|||
func sagaBarrierTransOut(c *gin.Context) (interface{}, error) { |
|||
gid := c.Query("gid") |
|||
lid := c.Query("lid") |
|||
req := reqFrom(c) |
|||
return dtm.ThroughBarrierCall(dbGet().ToSqlDB(), "saga", gid, lid, "action", func(sdb *sql.DB) (interface{}, error) { |
|||
db := common.SqlDB2DB(sdb) |
|||
dbr := db.Model(&UserAccount{}).Where("user_id = ?", c.Query("user_id")). |
|||
Update("balance", gorm.Expr("balance - ?", req.Amount)) |
|||
return nil, dbr.Error |
|||
}) |
|||
|
|||
// res := common.OrString(SagaBarrierTransOutResult, req.TransOutResult, "SUCCESS")
|
|||
// logrus.Printf("%s TransOut: %v result: %s", gid, req, res)
|
|||
// return M{"result": res}, nil
|
|||
} |
|||
|
|||
func sagaBarrierTransOutCompensate(c *gin.Context) (interface{}, error) { |
|||
gid := c.Query("gid") |
|||
req := reqFrom(c) |
|||
res := common.OrString(SagaBarrierTransOutCompensateResult, "SUCCESS") |
|||
logrus.Printf("%s TransOutCompensate: %v result: %s", gid, req, res) |
|||
return M{"result": res}, nil |
|||
} |
|||
@ -0,0 +1,134 @@ |
|||
# 子事务屏障--彻底解决子事务乱序执行问题 |
|||
### 子事务乱序问题 |
|||
我们拿分布式事务中的TCC作为例子，看看如何子事务乱序是什么样的，如何解决。 |
|||
假设一个分布式TCC事务G，包含子事务A和B，全部成功为 |
|||
A-Try ok -> B-Try ok -> G完成 |
|||
当网络出现问题时，下面是某一种常见情形 |
|||
A-Try 网络丢包导致超时 -> A Cancel ok丢包 -> A Try 请求到达子事务进行处理（此时因为Cancel已成功，需要忽略，立即返回)->A Cancel ok |
|||
这种情况下： |
|||
A Cancel在Try之前执行，称为空补偿问题，此时Cancel的处理，需要进行空补偿，称为空补偿控制 |
|||
A 第二次Cancel，称为幂等问题，此时Cancel的处理，需要判断此前已回滚，此次忽略，成为幂等控制 |
|||
A 第二次Try时，之前已进行了Cancel，此次之行应当忽略，我们称之为防悬挂控制 |
|||
这几种情况的正确处理，通常需要子事务做精细处理，例如记录业务处理的主键，并保证上述描述的逻辑。 |
|||
|
|||
子事务屏障提供了一套全新方法，让业务编写者完全不用被上述问题困扰，他的工作方式如下： |
|||
|
|||
子事务进入屏障，在屏障中编写自己的业务逻辑，由屏障保证内部Try-Confirm-Cancel逻辑只被提交一次 |
|||
|
|||
子事务屏障的原理如下： |
|||
它使用表sub_trans_barrier，主键为全局事务id-子事务id-子事务分支名称（try|confirm|cancel） |
|||
。开启事务 |
|||
。如果是Try分支，则那么insert ignore插入gid-branchid-try，如果成功插入，则调用屏障内逻辑 |
|||
. 如果是Confirm分支，那么insert ignore插入gid-branchid-confirm，如果成功插入，则调用屏障内逻辑 |
|||
. 如果是Cancel分支，那么insert ignore插入gid-branchid-try，再插入gid-branchid-cancel，如果try未插入并且cancel插入成功，则调用屏障内逻辑 |
|||
屏障内如果调用成功，提交事务，返回成功 |
|||
如果调用异常，回滚事务，返回异常 |
|||
|
|||
在此机制下，解决了乱序问题 |
|||
空补偿控制--如果Try没有执行，直接执行了Cancel，那么Cancel插入gid-branchid-try会成功，不走屏障内的逻辑，保证了空补偿控制 |
|||
幂等控制--任何一个分支都无法重复插入唯一键，保证了不会重复执行 |
|||
防悬挂控制--Try在Cancel之后执行，那么插入的gid-branchid-try不成功，就不执行，保证了防悬挂控制 |
|||
|
|||
通过子事务屏障，完全解决了子事务乱序问题，业务人员可以只关心自己的业务逻辑 |
|||
|
|||
|
|||
某些业务要求，一系列操作必须全部执行，而不能仅执行一部分。例如，一个转账操作： |
|||
|
|||
``` |
|||
-- 从id=1的账户给id=2的账户转账100元 |
|||
-- 第一步：将id=1的A账户余额减去100 |
|||
UPDATE accounts SET balance = balance - 100 WHERE id = 1; |
|||
-- 第二步：将id=2的B账户余额加上100 |
|||
UPDATE accounts SET balance = balance + 100 WHERE id = 2; |
|||
``` |
|||
这两条SQL语句必须全部执行，或者，由于某些原因，如果第一条语句成功，第二条语句失败，就必须全部撤销。 |
|||
|
|||
这种把多条语句作为一个整体进行操作的功能，被称为数据库事务。数据库事务可以确保该事务范围内的所有操作都可以全部成功或者全部失败。如果事务失败，那么效果就和没有执行这些SQL一样，不会对数据库数据有任何改动。 |
|||
|
|||
[更多事务介绍](https://www.liaoxuefeng.com/wiki/1177760294764384/1179611198786848) |
|||
|
|||
|
|||
### 微服务 |
|||
|
|||
如果一个事务涉及的所有操作能够放在一个服务内部，那么使用各门语言里事务相关的库，可以轻松的实现多个操作作为整体的事务操作。 |
|||
|
|||
但是有些服务，例如生成订单涉及做很多操作，包括库存、优惠券、赠送、账户余额等。当系统复杂程度增加时，想要把所有这些操作放到一个服务内实现，会导致耦合度太高，维护成本非常高。 |
|||
|
|||
针对复杂的系统，当前流行的微服务架构是非常好的解决方案，该架构能够把复杂系统进行拆分，拆分后形成了大量微服务，独立开发，独立维护。 |
|||
|
|||
[更多微服务介绍](https://www.zhihu.com/question/65502802) |
|||
|
|||
虽然服务拆分了，但是订单本身的逻辑需要多个操作作为一个整体，要么全部成功，要么全部失败，这就带来了新的挑战。如何把散落在各个微服务中的本地事务，组成一个大的事务，保证他们作为一个整体，这就是分布式事务需要解决的问题。 |
|||
|
|||
### 分布式事务 |
|||
分布式事务简单的说，就是一次大的操作由不同的小操作组成，这些小的操作分布在不同的服务器上，且属于不同的应用，分布式事务需要保证这些小操作要么全部成功，要么全部失败。本质上来说，分布式事务就是为了保证不同数据库的数据一致性。 |
|||
|
|||
[更多分布式事务介绍](https://juejin.cn/post/6844903647197806605) |
|||
|
|||
分布式事务方案包括： |
|||
* xa |
|||
* tcc |
|||
* saga |
|||
* 可靠消息 |
|||
|
|||
下面我们看看最简单的xa |
|||
|
|||
### XA |
|||
|
|||
XA一共分为两阶段： |
|||
|
|||
第一阶段（prepare）：事务管理器向所有本地资源管理器发起请求，询问是否是 ready 状态，所有参与者都将本事务能否成功的信息反馈发给协调者； |
|||
|
|||
第二阶段 (commit/rollback)：事务管理器根据所有本地资源管理器的反馈，通知所有本地资源管理器，步调一致地在所有分支上提交或者回滚。 |
|||
|
|||
目前主流的数据库基本都支持XA事务，包括mysql、oracle、sqlserver、postgre |
|||
|
|||
### xa实践 |
|||
|
|||
介绍了这么多，我们来实践完成一个微服务上的xa事务，加深分布式事务的理解，这里将采用[dtm](https://github.com/yedf/dtm.git)作为示例 |
|||
|
|||
[安装go](https://golang.org/doc/install) |
|||
|
|||
[安装mysql](https://www.mysql.com/cn/) |
|||
|
|||
获取dtm |
|||
``` |
|||
git clone https://github.com/yedf/dtm.git |
|||
cd dtm |
|||
``` |
|||
配置mysql |
|||
``` |
|||
cp conf.sample.yml conf.yml |
|||
vi conf.yml |
|||
``` |
|||
|
|||
运行示例 |
|||
|
|||
``` |
|||
go run app/main.go xa |
|||
``` |
|||
|
|||
从日志里，能够找到以下输出 |
|||
``` |
|||
# 服务1输出 |
|||
XA start '4fPqCNTYeSG' |
|||
UPDATE `user_account` SET `balance`=balance + 30,`update_time`='2021-06-09 11:50:42.438' WHERE user_id = '1' |
|||
XA end '4fPqCNTYeSG' |
|||
XA prepare '4fPqCNTYeSG' |
|||
|
|||
# 服务2输出 |
|||
XA start '4fPqCPijxyC' |
|||
UPDATE `user_account` SET `balance`=balance - 30,`update_time`='2021-06-09 11:50:42.493' WHERE user_id = '2' |
|||
XA end '4fPqCPijxyC' |
|||
XA prepare '4fPqCPijxyC' |
|||
|
|||
# 服务1输出 |
|||
xa commit '4fPqCNTYeSG' |
|||
|
|||
#服务2输出 |
|||
xa commit '4fPqCPijxyC' |
|||
``` |
|||
|
|||
|
|||
### 总结 |
|||
在这篇简短的文章里，我们大致介绍了 事务->分布式事务->微服务处理XA事务。有兴趣的同学可以通过[dtm](https://github.com/yedf/dtm)继续研究分布式事务 |
|||
Loading…
Reference in new issue