碼上歡樂
相關內容    簡體    繁體

帶你十天輕松搞定 Go 微服務之大結局（分布式事務）

本文轉載自   查看原文   2022-02-15 15:20   1007    微服務/ go-zero/ Go/ go/ web開發

序言

我們通過一個系列文章跟大家詳細展示一個 go-zero 微服務示例，整個系列分十篇文章，目錄結構如下：

  1. 環境搭建
  2. 服務拆分
  3. 用戶服務
  4. 產品服務
  5. 訂單服務
  6. 支付服務
  7. RPC 服務 Auth 驗證
  8. 服務監控
  9. 鏈路追蹤
  10. 分布式事務（本文）

期望通過本系列帶你在本機利用 Docker 環境利用 go-zero 快速開發一個商城系統，讓你快速上手微服務。

完整示例代碼：https://github.com/nivin-studio/go-zero-mall

首先，我們來看一下整體的服務拆分圖：

10.1 DTM 介紹

DTM 是一款 golang 開發的分布式事務管理器，解決了跨數據庫、跨服務、跨語言棧更新數據的一致性問題。

絕大多數的訂單系統的事務都會跨服務，因此都有更新數據一致性的需求，都可以通過 DTM 大幅簡化架構，形成一個優雅的解決方案。

而且 DTM 已經深度合作，原生的支持go-zero中的分布式事務，下面就來詳細的講解如何用 DTM 來幫助我們的訂單系統解決一致性問題

10.2 go-zero 使用 DTM

首先我們回顧下 第五章 訂單服務 中 order rpc 服務中 Create 接口處理邏輯。方法里判斷了用戶和產品的合法性，以及產品庫存是否充足，最后通過 OrderModel 創建了一個新的訂單，以及調用 product rpc 服務 Update 的接口更新了產品的庫存。

func (l *CreateLogic) Create(in *order.CreateRequest) (*order.CreateResponse, error) {
	// 查詢用戶是否存在
	_, err := l.svcCtx.UserRpc.UserInfo(l.ctx, &user.UserInfoRequest{
		Id: in.Uid,
	})
	if err != nil {
		return nil, err
	}

	// 查詢產品是否存在
	productRes, err := l.svcCtx.ProductRpc.Detail(l.ctx, &product.DetailRequest{
		Id: in.Pid,
	})
	if err != nil {
		return nil, err
	}
	// 判斷產品庫存是否充足
	if productRes.Stock <= 0 {
		return nil, status.Error(500, "產品庫存不足")
	}

	newOrder := model.Order{
		Uid:    in.Uid,
		Pid:    in.Pid,
		Amount: in.Amount,
		Status: 0,
	}

	res, err := l.svcCtx.OrderModel.Insert(&newOrder)
	if err != nil {
		return nil, status.Error(500, err.Error())
	}

	newOrder.Id, err = res.LastInsertId()
	if err != nil {
		return nil, status.Error(500, err.Error())
	}

	_, err = l.svcCtx.ProductRpc.Update(l.ctx, &product.UpdateRequest{
		Id:     productRes.Id,
		Name:   productRes.Name,
		Desc:   productRes.Desc,
		Stock:  productRes.Stock - 1,
		Amount: productRes.Amount,
		Status: productRes.Status,
	})
	if err != nil {
		return nil, err
	}

	return &order.CreateResponse{
		Id: newOrder.Id,
	}, nil
}

之前我們說過，這里處理邏輯存在數據一致性問題，有可能訂單創建成功了，但是在更新產品庫存的時候可能會發生失敗，這時候就會存在訂單創建成功，產品庫存沒有減少的情況。

因為這里的產品庫存更新是跨服務操作的，也沒有辦法使用本地事務來處理，所以我們需要使用分布式事務來處理它。這里我們需要借助 DTM 的 SAGA 協議來實現訂單創建和產品庫存更新的跨服務分布式事務操作。

大家可以先移步到 DTM 的文檔先了接下 SAGA事務模式。

10.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'

# ......

10.2.2 添加 dtm_barrier 數據表

微服務是一個分布式系統，因此可能發生各種異常，例如網絡抖動導致重復請求，這類的異常會讓業務處理異常復雜。而 DTM 中，首創了 子事務屏障 技術，使用該技術，能夠非常便捷的解決異常問題，極大的降低了分布式事務的使用門檻。

使用 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。

10.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.go

package 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)
}

10.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 工具重新生成下代碼。

  • 實現 DecrStock 接口方法

    在這里只有庫存不足時，我們不需要再重試，直接回滾。

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

package logic

import (
	"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/tal-tech/go-zero/core/logx"
	"github.com/tal-tech/go-zero/core/stores/sqlx"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
)

type DecrStockLogic struct {
	ctx    context.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 logic

import (
	"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/tal-tech/go-zero/core/logx"
	"github.com/tal-tech/go-zero/core/stores/sqlx"
	"google.golang.org/grpc/status"
)

type DecrStockRevertLogic struct {
	ctx    context.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
}

10.2.5 修改 order rpc 服務

  • 添加 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 工具重新生成下代碼。

  • 修改 Create 接口方法

    原來 Create 接口方法中產品庫存判斷和更新操作，我們已經在 product rpc DecrStock 接口方法中實現了，所以我們這里只要創建訂單一個操作即可。

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

package logic

import (
	"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/tal-tech/go-zero/core/logx"
	"github.com/tal-tech/go-zero/core/stores/sqlx"
	"google.golang.org/grpc/status"
)

type CreateLogic struct {
	ctx    context.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
}

  • 實現 CreateRevert 接口方法

    在這個接口中我們查詢用戶剛剛創建的訂單，把訂單的狀態改為 9（無效狀態）。

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

package logic

import (
	"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/tal-tech/go-zero/core/logx"
	"github.com/tal-tech/go-zero/core/stores/sqlx"
	"google.golang.org/grpc/status"
)

type CreateRevertLogic struct {
	ctx    context.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
}

10.2.6 修改 order api 服務

我們把 order rpc 服務 Create、CreateRevert 接口方法，product rpc 服務 DecrStock、DecrStockRevert 接口方法，提到 order api 服務中做成一個以 SAGA事務模式 的分布式事務操作。

  • 添加 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.rpc

ProductRpc:
  Etcd:
    Hosts:
    - etcd:2379
    Key: product.rpc
  • 添加 pproduct rpc 服務配置的實例化
$ vim mall/service/order/api/internal/config/config.go

package config

import (
	"github.com/tal-tech/go-zero/rest"
	"github.com/tal-tech/go-zero/zrpc"
)

type Config struct {
	rest.RestConf

	Auth struct {
		AccessSecret string
		AccessExpire int64
	}

	OrderRpc   zrpc.RpcClientConf
	ProductRpc zrpc.RpcClientConf
}
  • 注冊服務上下文 pproduct rpc 的依賴
$ vim mall/service/order/api/internal/svc/servicecontext.go

package svc

import (
	"mall/service/order/api/internal/config"
	"mall/service/order/rpc/orderclient"
	"mall/service/product/rpc/productclient"

	"github.com/tal-tech/go-zero/zrpc"
)

type ServiceContext struct {
	Config config.Config

	OrderRpc   orderclient.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)),
	}
}
  • 添加導入 gozero 的 dtm 驅動
$ vim mall/service/order/api/order.go

package main

import (
	......

	_ "github.com/dtm-labs/driver-gozero" // 添加導入 `gozero` 的 `dtm` 驅動
)

var configFile = flag.String("f", "etc/order.yaml", "the config file")

func main() {
	......
}
  • 修改 order api Create 接口方法
$ vim mall/service/order/api/internal/logic/createlogic.go

package logic

import (
	"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/tal-tech/go-zero/core/logx"
	"google.golang.org/grpc/status"
)

type CreateLogic struct {
	logx.Logger
	ctx    context.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 搜索即可找到。

10.3 測試 go-zero + DTM

10.3.1 測試分布式事務正常流程

  • 使用 postman 調用 /api/product/create 接口，創建一個產品，庫存 stock 為 1。

  • 使用 postman 調用 /api/order/create 接口，創建一個訂單，產品ID pid 為 1。

  • 我們可以看出，產品的庫存從原來的 1 已經變成了 0。

  • 我們再看下子事務屏障表 barrier 里的數據，我們可以看出兩個服務的操作均已經完成。

10.3.2 測試分布式事務失敗流程1

  • 接着上面測試結果，此時的產品ID為 1 的庫存已經是 0了, 使用 postman 調用 /api/order/create 接口，再創建一個訂單。

  • 我們看下訂單數據表里有一條ID為 2 產品ID為 1 的數據，它的訂單數據狀態為 9。

  • 我們再看下子事務屏障表 barrier 里的數據，我們可以看出（gid = fqYS8CbYbK8GkL8SCuTRUF）第一個服務（branch_id = 01）子事務屏障操作是正常，第二個服務（branch_id = 02）子事務屏障操作失敗，要求補償。於是兩個服務都發生了補償的操作記錄。

  • 這個分布式事務的操作流程

    1. 首先 DTM 服務會調 order rpc Create 接口進行創建訂單處理。
    2. 創建訂單完成后 DTM 服務再調 product rpc DecrStock 接口，這個接口的里通過 pid 更新產品庫存，因產品庫存不足，拋出事務失敗。
    3. DTM 服務發起補償機制，調 order rpc CreateRevert 接口進行訂單的補償處理。
    4. DTM 服務發起補償機制，調 product rpc DecrStockRevert 接口進行產品庫存更新的補償處理。但是因為在 product rpc DecrStock 接口的子事務屏障內，業務處理並未成功。所以在 DecrStockRevert 接口里不會執行子事務屏障內的業務邏輯。

10.3.3 測試分布式事務失敗流程2

  • 我們在數據庫中手動將產品ID為 1 庫存修改為100，然后在 product rpc DecrStock 接口方法中子事務屏障外，人為的制造異常失敗。

  • 使用 postman 調用 /api/order/create 接口，再創建一個訂單，產品ID pid 為 1。

  • 我們分別來看下訂單數據表和產品數據表，訂單數據表ID為 3 的訂單，它的訂單數據狀態為 9。產品數據表ID為 1 的產品，它的庫存還是 100 且數據更新時間也發生了變化。

  • 我們再看下子事務屏障表 barrier 里的數據，我們可以看出（gid = ZbjYHv2jNra7RMwyWjB5Lc）第一個服務（branch_id = 01）子事務屏障操作是正常，第二個服務（branch_id = 02）子事務屏障操作也是正常。因為在 product rpc DecrStock 接口方法中子事務屏障外，我們人為的制造異常失敗，所以兩個服務發生了補償的操作記錄。

大家可以對比下 測試分布式事務失敗流程1 與 測試分布式事務失敗流程2 不同之處，是不是能發現和體會到 DTM 的這個子事務屏障技術的強大之處。

子事務屏障會自動識別正向操作是否已執行，失敗流程1未執行業務操作，所以補償時，也不會執行補償的業務操作；失敗流程2執行了業務操作，所以補償時，也會執行補償的業務操作。

項目地址

https://github.com/zeromicro/go-zero

歡迎使用 go-zero 並 star 支持我們！

微信交流群

關注『微服務實踐』公眾號並點擊 交流群 獲取社區群二維碼。


免責聲明！

本站轉載的文章為個人學習借鑒使用，本站對版權不負任何法律責任。如果侵犯了您的隱私權益，請聯系本站郵箱yoyou2525@163.com刪除。



猜您在找 帶你十天輕松搞定 Go 微服務系列（一） 帶你十天輕松搞定 Go 微服務系列（三） 帶你十天輕松搞定 Go 微服務系列（五） 帶你十天輕松搞定 Go 微服務系列（七） 帶你十天輕松搞定 Go 微服務系列（二） 帶你十天輕松搞定 Go 微服務系列（四） 帶你十天輕松搞定 Go 微服務系列（八、服務監控） 帶你十天輕松搞定 Go 微服務系列（九、鏈路追蹤） PHP 微服務之【分布式事務】 輕松構建微服務之分布式配置中心
 
粵ICP備18138465號   © 2018-2025 CODEPRJ.COM