源碼:point_down: seata-golang

概述

我們知道 Seata Java Client 的 AT 模式，通過代理數(shù)據(jù)源，實現(xiàn)了對業(yè)務(wù)代碼無侵入的分布式事務(wù)協(xié)調(diào)機制，將與 Transaction Coordinator (TC) 交互的邏輯、Commit 的邏輯、Rollback 的邏輯，隱藏在切面和代理數(shù)據(jù)源相應(yīng)的代碼中，使開發(fā)者無感知。那如果這個方法，要用 Golang 來實現(xiàn)一遍，應(yīng)該如何操作呢？關(guān)于這個問題，我想了很久，最初的設(shè)想是，對 database/sql 的 mysql driver 進行增強，在對包 github.com/go-sql-driver/mysql 研究了一段時間后，還是沒有頭緒，不知如何下手，最后轉(zhuǎn)而增強 database/sql 包。由于 AT 模式必須保證本地事務(wù)的正確處理，在具體業(yè)務(wù)開發(fā)時，首先要通過 db.Begin() 獲得一個 Tx 對象，然后再 tx.Exec() 執(zhí)行數(shù)據(jù)庫操作，最后 tx.Commit() 提交或 tx.Rollback() 回滾。這種處理方式算是一個 Golang 數(shù)據(jù)庫事務(wù)處理的基本操作。 所以對 database/sql 的增強，我們重點關(guān)注這幾個方法 db.Begin() 、 tx.Exec() 、 tx.Commit() 、 tx.Rollback 。

事務(wù)提交、回滾

通過 Seata Java Client 的相關(guān)代碼，我們知道，在本地事務(wù)提交的時候，主要是將分支事務(wù)注冊到 TC 上，并將數(shù)據(jù)庫操作產(chǎn)生的 undoLog 一起寫入到 undoLog 表；本地事務(wù)回滾的時候，需要將分支事務(wù)(即本地事務(wù))的執(zhí)行狀態(tài)報告給 TC，使 TC 好知道是否通知參與全局事務(wù)的其他分支回滾。

func (tx *Tx) Commit() error { //注冊分支事務(wù)branchId,err := tx.register()if err != nil {return errors.WithStack(err)}tx.tx.Context.BranchId = branchIdif tx.tx.Context.HasUndoLog() { //將 undoLog 寫入 undoLog 表err = manager.GetUndoLogManager().FlushUndoLogs(tx.tx)if err != nil {err1 := tx.report(false)if err1 != nil {return errors.WithStack(err1)}return errors.WithStack(err)}err = tx.tx.Commit()if err != nil {err1 := tx.report(false)if err1 != nil {return errors.WithStack(err1)}return errors.WithStack(err)}} else {return tx.tx.Commit()}if tx.reportSuccessEnable {tx.report(true)}tx.tx.Context.Reset()return nil}

db.Begin() 會產(chǎn)生一個 Tx 對象， tx.Exec() 會產(chǎn)生 undoLog， tx.Commit() 將 undoLog 刷到數(shù)據(jù)庫中。那么 undoLog 保存到哪里呢？答案是 Tx_Context 中。

type TxContext struct {*context.RootContextXid stringBranchId int64IsGlobalLockRequire boolLockKeysBuffer *model.SetSqlUndoItemsBuffer []*undo.SqlUndoLog}

Commit() 方法中的 tx.tx.Context ，第一個 tx 是封裝的 Tx 對象，第二個 tx 是 database/sql 的 Tx， tx.tx.Context 則是 Tx_Contex。UndoLogManager 則是操作 undoLog 的核心對象，處理 undoLog 的插入、刪除，并查詢出 undoLog 用于回滾。

func (tx *Tx) Rollback() error {err := tx.tx.Rollback()if tx.tx.Context.InGlobalTransaction() && tx.tx.Context.IsBranchRegistered() { // 報告 TC 分支事務(wù)執(zhí)行失敗tx.report(false)}tx.tx.Context.Reset()return err}

通過上面的代碼呢，我們知道增強型 Tx 對象需要向 TC 注冊分支事務(wù)，并報告分支事務(wù)的執(zhí)行狀態(tài)，相應(yīng)代碼如下：

func (tx *Tx) register() (int64,error) {return dataSourceManager.BranchRegister(meta.BranchTypeAT,tx.tx.ResourceId,"",tx.tx.Context.Xid,nil,tx.tx.Context.BuildLockKeys())}func (tx *Tx) report(commitDone bool) error {retry := tx.reportRetryCountfor retry > 0 {var err errorif commitDone {err = dataSourceManager.BranchReport(meta.BranchTypeAT, tx.tx.Context.Xid, tx.tx.Context.BranchId,meta.BranchStatusPhaseoneDone,nil)} else {err = dataSourceManager.BranchReport(meta.BranchTypeAT, tx.tx.Context.Xid, tx.tx.Context.BranchId,meta.BranchStatusPhaseoneFailed,nil)}if err != nil {logging.Logger.Errorf("Failed to report [%d/%s] commit done [%t] Retry Countdown: %d",tx.tx.Context.BranchId,tx.tx.Context.Xid,commitDone,retry)retry = retry -1if retry == 0 {return errors.WithMessagef(err,"Failed to report branch status %t",commitDone)}}}return nil}

和 TC 進行通信的主要邏輯還是在 DataSourceManager 里面。AT 模式涉及的兩個關(guān)鍵對象 DataSourceManager、UndoLogManager 就浮出水面。一個用于遠程 TC 交互，一個用于本地數(shù)據(jù)庫處理。

事務(wù)執(zhí)行

func (tx *Tx) Exec(query string, args ...interface{}) (sql.Result, error) {var parser = p.New() // 解析業(yè)務(wù) sqlact,_ := parser.ParseOneStmt(query,"","")deleteStmt,isDelete := act.(*ast.DeleteStmt)if isDelete {executor := &DeleteExecutor{tx: tx.tx,sqlRecognizer: mysql.NewMysqlDeleteRecognizer(query,deleteStmt),values: args,}return executor.Execute()}insertStmt,isInsert := act.(*ast.InsertStmt)if isInsert {executor := &InsertExecutor{tx: tx.tx,sqlRecognizer: mysql.NewMysqlInsertRecognizer(query,insertStmt),values: args,}return executor.Execute()}updateStmt,isUpdate := act.(*ast.UpdateStmt)if isUpdate {executor := &UpdateExecutor{tx: tx.tx,sqlRecognizer: mysql.NewMysqlUpdateRecognizer(query,updateStmt),values: args,}return executor.Execute()}return tx.tx.Tx.Exec(query,args)}

執(zhí)行業(yè)務(wù) sql，并生成 undoLog 的關(guān)鍵，在于識別業(yè)務(wù) sql 執(zhí)行了什么操作：插入？刪除？修改？這里使用 tidb 的 sql parser 去解析業(yè)務(wù) sql，再使用相應(yīng)的執(zhí)行器去執(zhí)行業(yè)務(wù) sql，生成 undoLog 保存在 Tx_Context 中。

事務(wù)開啟

db.Begin() 返回增強型的 Tx 對象。

func (db *DB) Begin(ctx *context.RootContext) (*Tx,error) {tx,err := db.DB.Begin()if err != nil {return nil,err}proxyTx := &tx2.ProxyTx{Tx: tx,DSN: db.conf.DSN,ResourceId: db.GetResourceId(),Context: tx2.NewTxContext(ctx),}return &Tx{tx: proxyTx,reportRetryCount: db.conf.ReportRetryCount,reportSuccessEnable: db.conf.ReportSuccessEnable,},nil}

seata-golang at 模式的使用

sample 代碼

• 首先執(zhí)行 scripts 腳本，初始化數(shù)據(jù)庫
  如果之前沒有初始化過 seata 數(shù)據(jù)庫，先執(zhí)行 seata-golang/scripts/server/db/mysql.sql 腳本
• 修改 dsn 數(shù)據(jù)庫配置，修改下列文件：

seata-golang/tc/app/profiles/dev/config.ymlseata-golang/samples/at/product_svc/conf/client.ymlseata-golang/samples/at/product_svc/conf/client.yml

• 將下列文件中的 configPath 修改為 client.yml 配置文件的路徑

seata-golang/samples/at/product_svc/main.goseata-golang/samples/at/order_svc/main.goseata-golang/samples/at/aggregation_svc/main.go

• 依次運行 tc、order_svc、product_svc、aggragation_svc，訪問下列地址開始測試：

http://localhost:8003/createSoCommithttp://localhost:8003/createSoRollback

總結(jié)

以上是生活随笔為你收集整理的属于db模式缺点的是什么_详解 Seata Golang 客户端 AT 模式及其使用的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。