大连专业html5网站建设,企业网站建设解决方案,北京网站优化公司如何,网页制作教程哔哩源码:point_down: seata-golang概述我们知道 Seata Java Client 的 AT 模式#xff0c;通过代理数据源#xff0c;实现了对业务代码无侵入的分布式事务协调机制#xff0c;将与 Transaction Coordinator (TC) 交互的逻辑、Commit 的逻辑、Rollback 的逻辑#xff0c;隐藏在…源码:point_down: seata-golang概述我们知道 Seata Java Client 的 AT 模式通过代理数据源实现了对业务代码无侵入的分布式事务协调机制将与 Transaction Coordinator (TC) 交互的逻辑、Commit 的逻辑、Rollback 的逻辑隐藏在切面和代理数据源相应的代码中使开发者无感知。那如果这个方法要用 Golang 来实现一遍应该如何操作呢关于这个问题我想了很久最初的设想是对 database/sql 的 mysql driver 进行增强在对包 github.com/go-sql-driver/mysql 研究了一段时间后还是没有头绪不知如何下手最后转而增强 database/sql 包。由于 AT 模式必须保证本地事务的正确处理在具体业务开发时首先要通过 db.Begin() 获得一个 Tx 对象然后再 tx.Exec() 执行数据库操作最后 tx.Commit() 提交或 tx.Rollback() 回滚。这种处理方式算是一个 Golang 数据库事务处理的基本操作。 所以对 database/sql 的增强我们重点关注这几个方法 db.Begin() 、 tx.Exec() 、 tx.Commit() 、 tx.Rollback 。事务提交、回滚通过 Seata Java Client 的相关代码我们知道在本地事务提交的时候主要是将分支事务注册到 TC 上并将数据库操作产生的 undoLog 一起写入到 undoLog 表本地事务回滚的时候需要将分支事务(即本地事务)的执行状态报告给 TC使 TC 好知道是否通知参与全局事务的其他分支回滚。func (tx *Tx) Commit() error { //注册分支事务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() 会产生一个 Tx 对象 tx.Exec() 会产生 undoLog tx.Commit() 将 undoLog 刷到数据库中。那么 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 分支事务执行失败tx.report(false)}tx.tx.Context.Reset()return err}通过上面的代码呢我们知道增强型 Tx 对象需要向 TC 注册分支事务并报告分支事务的执行状态相应代码如下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 模式涉及的两个关键对象 DataSourceManager、UndoLogManager 就浮出水面。一个用于远程 TC 交互一个用于本地数据库处理。事务执行func (tx *Tx) Exec(query string, args ...interface{}) (sql.Result, error) {var parser p.New() // 解析业务 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)}执行业务 sql并生成 undoLog 的关键在于识别业务 sql 执行了什么操作插入删除修改这里使用 tidb 的 sql parser 去解析业务 sql再使用相应的执行器去执行业务 sql生成 undoLog 保存在 Tx_Context 中。事务开启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 代码首先执行 scripts 脚本初始化数据库如果之前没有初始化过 seata 数据库先执行 seata-golang/scripts/server/db/mysql.sql 脚本修改 dsn 数据库配置修改下列文件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
