1、学习课程 带你十天轻松搞定 Go 微服务系列（十）- 分布式事务
今天是学习 go 微服务的最后一天，今天是学习分布式事务
2、go-zero 使用 DTM 2.1 添加 DTM 服务配置 参见 第一章 环境搭建，修改 dtm->config.yml 配置文件。我们只要修改 MicroService 中的 Target，EndPoint 配置即可，将 dtm 注册到 etcd 中。

# ......# 微服务 MicroService: Driver: 'dtm-driver-gozero'# 要处理注册/发现的驱动程序的名称 Target: 'etcd://etcd:2379/dtmservice' # 注册 dtm 服务的 etcd 地址 EndPoint: 'dtm:36790'# ......

2.2 添加 dtm_barrier 数据表 使用 DTM 提供的子事务屏障技术则需要在业务数据库中创建子事务屏障相关的表，建表语句如下：

create database if not exists dtm_barrier /*!40100 DEFAULT CHARACTER SET utf8mb4 */ ; drop table if exists dtm_barrier.barrier; create table if not exists dtm_barrier.barrier( id bigint(22) PRIMARY KEY AUTO_INCREMENT, trans_type varchar(45) default '', gid varchar(128) default '', branch_id varchar(128) default '', op varchar(45) default '', barrier_id varchar(45) default '', reason varchar(45) default '' comment 'the branch type who insert this record', create_time datetime DEFAULT now(), update_time datetime DEFAULT now(), key(create_time), key(update_time), UNIQUE key(gid, branch_id, op, barrier_id) );

注意：库名和表名请勿修改，如果您自定义了表名，请在使用前调用 dtmcli.SetBarrierTableName。
2.3 修改 OrderModel 和 ProductModel 在每一个子事务中，很多操作逻辑，需要使用到本地事务，所以我们添加一些 model 方法兼容 DTM 的子事务屏障

$ vim mall/service/order/model/ordermodel.go

package model......type ( OrderModel interface { TxInsert(tx *sql.Tx, data *Order) (sql.Result, error) TxUpdate(tx *sql.Tx, data *Order) error FindOneByUid(uid int64) (*Order, error) } )......func (m *defaultOrderModel) TxInsert(tx *sql.Tx, data *Order) (sql.Result, error) { query := fmt.Sprintf("insert into %s (%s) values (?, ?, ?, ?)", m.table, orderRowsExpectAutoSet) ret, err := tx.Exec(query, data.Uid, data.Pid, data.Amount, data.Status) return ret, err }func (m *defaultOrderModel) TxUpdate(tx *sql.Tx, data *Order) error { productIdKey := fmt.Sprintf("%s%v", cacheOrderIdPrefix, data.Id) _, err := m.Exec(func(conn sqlx.SqlConn) (result sql.Result, err error) { query := fmt.Sprintf("update %s set %s where `id` = ?", m.table, orderRowsWithPlaceHolder) return tx.Exec(query, data.Uid, data.Pid, data.Amount, data.Status, data.Id) }, productIdKey) return err }func (m *defaultOrderModel) FindOneByUid(uid int64) (*Order, error) { var resp Order query := fmt.Sprintf("select %s from %s where `uid` = ? order by create_time desc limit 1", orderRows, m.table) err := m.QueryRowNoCache(&resp, query, uid) switch err { case nil: return &resp, nil case sqlc.ErrNotFound: return nil, ErrNotFound default: return nil, err } }$ vim mall/service/product/model/productmodel.gopackage model......type ( ProductModel interface { TxAdjustStock(tx *sql.Tx, id int64, delta int) (sql.Result, error) } )......func (m *defaultProductModel) TxAdjustStock(tx *sql.Tx, id int64, delta int) (sql.Result, error) { productIdKey := fmt.Sprintf("%s%v", cacheProductIdPrefix, id) return m.Exec(func(conn sqlx.SqlConn) (result sql.Result, err error) { query := fmt.Sprintf("update %s set stock=stock+? where stock >= -? and id=?", m.table) return tx.Exec(query, delta, delta, id) }, productIdKey) }

2.4 修改 product rpc 服务 添加 DecrStock, DecrStockRevert 接口方法
我们需要为 product rpc 服务添加 DecrStock、DecrStockRevert 两个接口方法，分别用于 产品库存更新 和 产品库存更新的补偿。

$ vim mall/service/product/rpc/product.proto

syntax = "proto3"; package productclient; option go_package = "product"; ......// 减产品库存 message DecrStockRequest { int64 id = 1; int64 num = 2; } message DecrStockResponse { } // 减产品库存service Product { ...... rpc DecrStock(DecrStockRequest) returns(DecrStockResponse); rpc DecrStockRevert(DecrStockRequest) returns(DecrStockResponse); }

提示：修改后使用 goctl 工具重新生成下代码。 要在 golang 容器下运行，可参考：Linux下部署go-zero，运行goctl model运行模板生成命令报错解决方法

$ cd mall/service/product $ goctl rpc proto -src ./rpc/product.proto -dir ./rpc

实现 DecrStock 接口方法
在这里只有库存不足时，我们不需要再重试，直接回滚。

$ vim mall/service/product/rpc/internal/logic/decrstocklogic.go

package logicimport ( "context" "database/sql" "mall/service/product/rpc/internal/svc" "mall/service/product/rpc/product" "github.com/dtm-labs/dtmcli" "github.com/dtm-labs/dtmgrpc" "github.com/zeromicro/go-zero/core/logx" "github.com/zeromicro/go-zero/core/stores/sqlx" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" )type DecrStockLogic struct { ctxcontext.Context svcCtx *svc.ServiceContext logx.Logger }func NewDecrStockLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DecrStockLogic { return &DecrStockLogic{ ctx:ctx, svcCtx: svcCtx, Logger: logx.WithContext(ctx), } }func (l *DecrStockLogic) DecrStock(in *product.DecrStockRequest) (*product.DecrStockResponse, error) { // 获取 RawDB db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB() if err != nil { return nil, status.Error(500, err.Error()) } // 获取子事务屏障对象 barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx) if err != nil { return nil, status.Error(500, err.Error()) } // 开启子事务屏障 err = barrier.CallWithDB(db, func(tx *sql.Tx) error { // 更新产品库存 result, err := l.svcCtx.ProductModel.TxAdjustStock(tx, in.Id, -1) if err != nil { return err }affected, err := result.RowsAffected() // 库存不足，返回子事务失败 if err == nil && affected == 0 { return dtmcli.ErrFailure }return err }) // 这种情况是库存不足，不再重试，走回滚 if err == dtmcli.ErrFailure { return nil, status.Error(codes.Aborted, dtmcli.ResultFailure) } if err != nil { return nil, err } return &product.DecrStockResponse{}, nil }

实现 DecrStockRevert 接口方法
在 DecrStock 接口方法中，产品库存是减去指定的数量，在这里我们把它给加回来。这样产品库存就回到在 DecrStock 接口方法减去之前的数量。

$ vim mall/service/product/rpc/internal/logic/decrstockrevertlogic.go

package logicimport ( "context" "database/sql" "mall/service/product/rpc/internal/svc" "mall/service/product/rpc/product" "github.com/dtm-labs/dtmcli" "github.com/dtm-labs/dtmgrpc" "github.com/zeromicro/go-zero/core/logx" "github.com/zeromicro/go-zero/core/stores/sqlx" "google.golang.org/grpc/status" )type DecrStockRevertLogic struct { ctxcontext.Context svcCtx *svc.ServiceContext logx.Logger }func NewDecrStockRevertLogic(ctx context.Context, svcCtx *svc.ServiceContext) *DecrStockRevertLogic { return &DecrStockRevertLogic{ ctx:ctx, svcCtx: svcCtx, Logger: logx.WithContext(ctx), } }func (l *DecrStockRevertLogic) DecrStockRevert(in *product.DecrStockRequest) (*product.DecrStockResponse, error) { // 获取 RawDB db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB() if err != nil { return nil, status.Error(500, err.Error()) } // 获取子事务屏障对象 barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx) if err != nil { return nil, status.Error(500, err.Error()) } // 开启子事务屏障 err = barrier.CallWithDB(db, func(tx *sql.Tx) error { // 更新产品库存 _, err := l.svcCtx.ProductModel.TxAdjustStock(tx, in.Id, 1) return err }) if err != nil { return nil, err } return &product.DecrStockResponse{}, nil }

2.5 修改 order rpc 服务 2.5.1 添加 CreateRevert 接口方法
order rpc 服务中已经有 Create 接口方法、我们需要创建它的补偿接口方法 CreateRevert。

$ vim mall/service/order/rpc/order.proto

syntax = "proto3"; package orderclient; option go_package = "order"; ......service Order { rpc Create(CreateRequest) returns(CreateResponse); rpc CreateRevert(CreateRequest) returns(CreateResponse); ...... }

提示：修改后使用 goctl 工具重新生成下代码。 也要在 golang 容器下运行。

$ cd mall/service/order $ goctl rpc proto -src ./rpc/order.proto -dir ./rpc

2.5.2 修改 Create 接口方法
原来 Create 接口方法中产品库存判断和更新操作，我们已经在 product rpcDecrStock 接口方法中实现了，所以我们这里只要创建订单一个操作即可。

$ vim mall/service/order/rpc/internal/logic/createlogic.go

package logicimport ( "context" "database/sql" "fmt" "mall/service/order/model" "mall/service/order/rpc/internal/svc" "mall/service/order/rpc/order" "mall/service/user/rpc/user" "github.com/dtm-labs/dtmgrpc" "github.com/zeromicro/go-zero/core/logx" "github.com/zeromicro/go-zero/core/stores/sqlx" "google.golang.org/grpc/status" )type CreateLogic struct { ctxcontext.Context svcCtx *svc.ServiceContext logx.Logger }func NewCreateLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateLogic { return &CreateLogic{ ctx:ctx, svcCtx: svcCtx, Logger: logx.WithContext(ctx), } }func (l *CreateLogic) Create(in *order.CreateRequest) (*order.CreateResponse, error) { // 获取 RawDB db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB() if err != nil { return nil, status.Error(500, err.Error()) } // 获取子事务屏障对象 barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx) if err != nil { return nil, status.Error(500, err.Error()) } // 开启子事务屏障 if err := barrier.CallWithDB(db, func(tx *sql.Tx) error { // 查询用户是否存在 _, err := l.svcCtx.UserRpc.UserInfo(l.ctx, &user.UserInfoRequest{ Id: in.Uid, }) if err != nil { return fmt.Errorf("用户不存在") }newOrder := model.Order{ Uid:in.Uid, Pid:in.Pid, Amount: in.Amount, Status: 0, } // 创建订单 _, err = l.svcCtx.OrderModel.TxInsert(tx, &newOrder) if err != nil { return fmt.Errorf("订单创建失败") }return nil }); err != nil { return nil, status.Error(500, err.Error()) } return &order.CreateResponse{}, nil }

2.5.3 实现 CreateRevert 接口方法
在这个接口中我们查询用户刚刚创建的订单，把订单的状态改为 9（无效状态）。

$ vim mall/service/order/rpc/internal/logic/createrevertlogic.go

package logicimport ( "context" "database/sql" "fmt" "mall/service/order/rpc/internal/svc" "mall/service/order/rpc/order" "mall/service/user/rpc/user" "github.com/dtm-labs/dtmgrpc" "github.com/zeromicro/go-zero/core/logx" "github.com/zeromicro/go-zero/core/stores/sqlx" "google.golang.org/grpc/status" )type CreateRevertLogic struct { ctxcontext.Context svcCtx *svc.ServiceContext logx.Logger }func NewCreateRevertLogic(ctx context.Context, svcCtx *svc.ServiceContext) *CreateRevertLogic { return &CreateRevertLogic{ ctx:ctx, svcCtx: svcCtx, Logger: logx.WithContext(ctx), } }func (l *CreateRevertLogic) CreateRevert(in *order.CreateRequest) (*order.CreateResponse, error) { // 获取 RawDB db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB() if err != nil { return nil, status.Error(500, err.Error()) } // 获取子事务屏障对象 barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx) if err != nil { return nil, status.Error(500, err.Error()) } // 开启子事务屏障 if err := barrier.CallWithDB(db, func(tx *sql.Tx) error { // 查询用户是否存在 _, err := l.svcCtx.UserRpc.UserInfo(l.ctx, &user.UserInfoRequest{ Id: in.Uid, }) if err != nil { return fmt.Errorf("用户不存在") } // 查询用户最新创建的订单 resOrder, err := l.svcCtx.OrderModel.FindOneByUid(in.Uid) if err != nil { return fmt.Errorf("订单不存在") } // 修改订单状态9，标识订单已失效，并更新订单 resOrder.Status = 9 err = l.svcCtx.OrderModel.TxUpdate(tx, resOrder) if err != nil { return fmt.Errorf("订单更新失败") }return nil }); err != nil { return nil, status.Error(500, err.Error()) } return &order.CreateResponse{}, nil }

2.6 修改 order api 服务 我们把 order rpc 服务 Create、CreateRevert 接口方法，product rpc 服务 DecrStock、DecrStockRevert 接口方法，提到 order api 服务中做成一个以 SAGA事务模式 的分布式事务操作。
2.6.1 添加 pproduct rpc 依赖配置

$ vim mall/service/order/api/etc/order.yaml

Name: Order Host: 0.0.0.0 Port: 8002......OrderRpc: Etcd: Hosts: - etcd:2379 Key: order.rpcProductRpc: Etcd: Hosts: - etcd:2379 Key: product.rpc

2.6.2 添加 pproduct rpc 服务配置的实例化

$ vim mall/service/order/api/internal/config/config.go

package configimport ( "github.com/zeromicro/go-zero/rest" "github.com/zeromicro/go-zero/zrpc" )type Config struct { rest.RestConf Auth struct { AccessSecret string AccessExpire int64 } OrderRpczrpc.RpcClientConf ProductRpc zrpc.RpcClientConf }

2.6.3 注册服务上下文 pproduct rpc 的依赖

$ vim mall/service/order/api/internal/svc/servicecontext.go

package svcimport ( "mall/service/order/api/internal/config" "mall/service/order/rpc/orderclient" "mall/service/product/rpc/productclient" "github.com/zeromicro/go-zero/zrpc" )type ServiceContext struct { Config config.Config OrderRpcorderclient.Order ProductRpc productclient.Product }func NewServiceContext(c config.Config) *ServiceContext { return &ServiceContext{ Config:c, OrderRpc:orderclient.NewOrder(zrpc.MustNewClient(c.OrderRpc)), ProductRpc: productclient.NewProduct(zrpc.MustNewClient(c.ProductRpc)), } }

2.6.4 添加导入 gozero 的 dtm 驱动

$ vim mall/service/order/api/order.go

package mainimport ( ...... _ "github.com/dtm-labs/driver-gozero" // 添加导入 `gozero` 的 `dtm` 驱动 )var configFile = flag.String("f", "etc/order.yaml", "the config file")func main() { ...... }

2.6.5 修改 order apiCreate 接口方法

$ vim mall/service/order/api/internal/logic/createlogic.go

package logicimport ( "context" "mall/service/order/api/internal/svc" "mall/service/order/api/internal/types" "mall/service/order/rpc/order" "mall/service/product/rpc/product" "github.com/dtm-labs/dtmgrpc" "github.com/zeromicro/go-zero/core/logx" "google.golang.org/grpc/status" )type CreateLogic struct { logx.Logger ctxcontext.Context svcCtx *svc.ServiceContext }func NewCreateLogic(ctx context.Context, svcCtx *svc.ServiceContext) CreateLogic { return CreateLogic{ Logger: logx.WithContext(ctx), ctx:ctx, svcCtx: svcCtx, } }func (l *CreateLogic) Create(req types.CreateRequest) (resp *types.CreateResponse, err error) { // 获取 OrderRpc BuildTarget orderRpcBusiServer, err := l.svcCtx.Config.OrderRpc.BuildTarget() if err != nil { return nil, status.Error(100, "订单创建异常") } // 获取 ProductRpc BuildTarget productRpcBusiServer, err := l.svcCtx.Config.ProductRpc.BuildTarget() if err != nil { return nil, status.Error(100, "订单创建异常") } // dtm 服务的 etcd 注册地址 var dtmServer = "etcd://etcd:2379/dtmservice" // 创建一个gid gid := dtmgrpc.MustGenGid(dtmServer) // 创建一个saga协议的事务 saga := dtmgrpc.NewSagaGrpc(dtmServer, gid). Add(orderRpcBusiServer+"/orderclient.Order/Create", orderRpcBusiServer+"/orderclient.Order/CreateRevert", &order.CreateRequest{ Uid:req.Uid, Pid:req.Pid, Amount: req.Amount, Status: 0, }). Add(productRpcBusiServer+"/productclient.Product/DecrStock", productRpcBusiServer+"/productclient.Product/DecrStockRevert", &product.DecrStockRequest{ Id:req.Pid, Num: 1, }) // 事务提交 err = saga.Submit() if err != nil { return nil, status.Error(500, err.Error()) } return &types.CreateResponse{}, nil }

• 提示：SagaGrpc.Add 方法第一个参数 action 是微服务 grpc 访问的方法路径，这个方法路径需要分别去以下文件中寻找。

mall/service/order/rpc/order/order.pb.go mall/service/product/rpc/product/product.pb.go

按关键字 Invoke 搜索即可找到。

文章图片

3 测试 go-zero + DTM 3.1 测试分布式事务正常流程 使用 postman 调用 /api/product/create 接口，创建一个产品，库存 stock 为 1。

文章图片

可是报错 401，no token present in request，这是因为创建商品的时候没有自动带登录的时候的 token，没有权限创建

文章图片

解决方式可以参考：新版Postman设置所有请求都自动带token登陆权限验证（Postman 版本9.10.0 ）
使用 postman 调用 /api/order/create 接口，创建一个订单，记得和上面一样设置请求自动带token登陆权限验证，产品ID pid 为 1。

文章图片


文章图片

我们可以看出，产品的库存从原来的 1 已经变成了 0。

文章图片

我们再看下子事务屏障表 barrier 里的数据，我们可以看出两个服务的操作均已经完成。

文章图片

4、系列 【学习笔记（带你十天轻松搞定|学习笔记：带你十天轻松搞定 Go 微服务系列大结局（十）- 分布式事务）】学习笔记：带你十天轻松搞定 Go 微服务系列（一）- 环境搭建
学习笔记：带你十天轻松搞定 Go 微服务系列（二）- 服务拆分
学习笔记：带你十天轻松搞定 Go 微服务系列（三）- 用户服务
学习笔记：带你十天轻松搞定 Go 微服务系列（四）- 产品服务
学习笔记：带你十天轻松搞定 Go 微服务系列（五）- 订单服务
学习笔记：带你十天轻松搞定 Go 微服务系列（六）- 支付服务
学习笔记：带你十天轻松搞定 Go 微服务系列（七）- RPC 服务 Auth 验证
学习笔记：带你十天轻松搞定 Go 微服务系列（八）- 服务监控
学习笔记：带你十天轻松搞定 Go 微服务系列（九）- 链路追踪
学习笔记：带你十天轻松搞定 Go 微服务系列大结局（十）- 分布式事务

推荐阅读

• 

  电视剧《七月与安生》中苏家明喜欢的到底是七月还是安生？
• 

  有没有什么可以让牙齿迅速变白的方法？
• 

  不锈钢管生产厂家有哪些 不锈钢管材
• 

  藏獒过剩 流浪藏獒会破坏高原生态雪豹会不会被捕食
• 

  2022-2023独生子女退休补贴新规 独生子女父母一次性补贴如何领取？
• 

  mongodb数据导入 同步mongodb到内存
• 

  linux系统更新内核,亲身体验 升级Linux系统内核完全过程
• 

  9种|9种 MySQL数据库优化的技巧
• 

  阿拉伯王子成名的机会或重现
• 

  爱普生供墨清零软件
• 

  药丸形|曝料：iPhone 14 Pro概念渲染图！“感叹号”挖孔屏
• 

  excel排序让后面的数据跟着动教程
• 

  山东省综合素质评价登录入口手机版 山东省综合素质评价登录入口
• 

  毛衣巧处理再穿不会痒 毛衣的日常护理法则
• 

  献给电脑族的呵护肌肤的7大方法
• 

  德国即热式电热水器,你知道什么原因吗
• 

  华润燃气热水器使用方法,不清楚的真可以看看
• 

  月饼的馅料有哪些 传统月饼的馅料有哪些
• 

  上等人 马龙等人如何处罚，下等人
• 

  运动塑身减肥的方法有哪些视频 运动塑身减肥的方法有哪些

• 学习Spring源码篇之环境搭建
• 背包四讲|背包四讲 (AcWing算法基础课笔记整理)
• 机器学习|河南郑州二手房房价预测
• Android安卓进阶之——一文带你了解抓包和反抓包
• 笔记|Jap技术总结
• rails实战
• 带你从内存的角度看Python中的变量
• Python全栈之学习CSS(2)
• 安全学习笔记-web安全之XSS攻击
• 三个点带你解读Smartbi商业智能大数据分析平台