preface
We will show you a go zero micro service example in detail through a series of articles. The whole series is divided into ten articles, and the directory structure is as follows:
- Environment construction
- Service splitting
- User services
- Product service
- Order service
- Payment services
- RPC service Auth authentication
- Service monitoring
- Link tracking
- Distributed transaction (this article)
Through this series, we hope to take you to quickly develop a mall system by using the Docker environment and go zero on the machine, so that you can quickly start micro services.
Complete sample code: github.com/nivin-studio/go-zero-ma...
First, let's take a look at the overall service splitting diagram:
10.1 introduction to DTM
DTM It is a distributed transaction manager developed by golang, which solves the consistency problem of updating data across databases, services and language stacks.
Most of the transactions of the order system will cross services, so there is a need to update data consistency. DTM can greatly simplify the architecture and form an elegant solution.
Moreover, DTM has deep cooperation and supports distributed transactions in go zero. Here is a detailed explanation of how to use DTM to help our order system solve the consistency problem
10.2 go zero uses DTM
First, let's review Chapter V order service Processing logic of Create interface in order rpc service in. Method determines the legitimacy of users and products and whether the product inventory is sufficient. Finally, a new order is created through OrderModel and the product inventory is updated by calling the interface of product rpc service Update.
func (l *CreateLogic) Create(in *order.CreateRequest) (*order.CreateResponse, error) {
// Query whether the user exists
_, err := l.svcCtx.UserRpc.UserInfo(l.ctx, &user.UserInfoRequest{
Id: in.Uid,
})
if err != nil {
return nil, err
}
// Query whether the product exists
productRes, err := l.svcCtx.ProductRpc.Detail(l.ctx, &product.DetailRequest{
Id: in.Pid,
})
if err != nil {
return nil, err
}
// Judge whether the product inventory is sufficient
if productRes.Stock <= 0 {
return nil, status.Error(500, "Insufficient product inventory")
}
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
}As we said before, there is a problem of data consistency in the processing logic here. It is possible that the order is created successfully, but it may fail when updating the product inventory. At this time, the order is created successfully and the product inventory is not reduced.
Because the product inventory update here is operated across services, and there is no way to use local transactions to process it, we need to use distributed transactions to process it. Here, we need the SAGA protocol of DTM to realize the cross service distributed transaction operation of order creation and product inventory update.
You can move to the DTM document first SAGA transaction mode.
10.2.1 add DTM service configuration
See Chapter 1 environment setup and modify dtm - > config YML configuration file. We just need to modify the Target and EndPoint configuration in MicroService and register dtm in etcd.
# ...... # Microservices MicroService: Driver: 'dtm-driver-gozero' # The name of the driver to process registration / discovery Target: 'etcd://etcd:2379/dtmservice '# register the etcd address of dtm service EndPoint: 'dtm:36790' # ......
10.2.2 add dtm_barrier data sheet
Microservice is a distributed system, so various exceptions may occur, such as repeated requests caused by network jitter. Such exceptions will make business processing extremely complex. In DTM, the first Sub transaction barrier Technology, the use of this technology can easily solve the abnormal problems, and greatly reduce the threshold of distributed transactions.
Using the sub transaction barrier technology provided by DTM, you need to create tables related to the sub transaction barrier in the business database. The table creation statements are as follows:
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) );
Note: do not modify the library name and table name. If you customize the name of the table, please call dtmcli. before use. SetBarrierTableName.
10.2.3 modify OrderModel and ProductModel
In each sub transaction, a lot of operation logic needs to be used for local transactions, so we add some model methods compatible with DTM sub transaction barriers
$ 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 modifying the product rpc service
Add the interface method of decrstock and decrstockrevert
We need to add two interface methods: DecrStock and DecrStockRevert to the product rpc service, which are used for product inventory update and compensation for product inventory update respectively.
$ vim mall/service/product/rpc/product.proto
syntax = "proto3";
package productclient;
option go_package = "product";
......
// Less product inventory
message DecrStockRequest {
int64 id = 1;
int64 num = 2;
}
message DecrStockResponse {
}
// Less product inventory
service Product {
......
rpc DecrStock(DecrStockRequest) returns(DecrStockResponse);
rpc DecrStockRevert(DecrStockRequest) returns(DecrStockResponse);
}Tip: after modification, use goctl tool to regenerate the next code.
Implementation of DecrStock interface method
Here, only when the inventory is insufficient, we do not need to retry and roll back directly.
$ 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) {
// Get RawDB
db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB()
if err != nil {
return nil, status.Error(500, err.Error())
}
// Get sub transaction barrier object
barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx)
if err != nil {
return nil, status.Error(500, err.Error())
}
// Open sub transaction barrier
err = barrier.CallWithDB(db, func(tx *sql.Tx) error {
// Update product inventory
result, err := l.svcCtx.ProductModel.TxAdjustStock(tx, in.Id, -1)
if err != nil {
return err
}
affected, err := result.RowsAffected()
// Insufficient inventory, failed to return sub transaction
if err == nil && affected == 0 {
return dtmcli.ErrFailure
}
return err
})
// In this case, the inventory is insufficient. Don't try again and go back
if err == dtmcli.ErrFailure {
return nil, status.Error(codes.Aborted, dtmcli.ResultFailure)
}
if err != nil {
return nil, err
}
return &product.DecrStockResponse{}, nil
}Implement the interface method of DecrStockRevert
In the DecrStock interface method, the product inventory is minus the specified quantity, and here we add it back. In this way, the product inventory returns to the previous quantity subtracted by the DecrStock interface method.
$ 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) {
// Get RawDB
db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB()
if err != nil {
return nil, status.Error(500, err.Error())
}
// Get sub transaction barrier object
barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx)
if err != nil {
return nil, status.Error(500, err.Error())
}
// Open sub transaction barrier
err = barrier.CallWithDB(db, func(tx *sql.Tx) error {
// Update product inventory
_, err := l.svcCtx.ProductModel.TxAdjustStock(tx, in.Id, 1)
return err
})
if err != nil {
return nil, err
}
return &product.DecrStockResponse{}, nil
}10.2.5 modify order rpc service
Add CreateRevert interface method
The order rpc service already has a Create interface method. We need to Create its compensation interface method 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);
......
}Tip: after modification, use goctl tool to regenerate the next code.
Modify Create interface method
In the original Create interface method, we have implemented the product inventory judgment and update operations in the product rpc DecrStock interface method, so we only need to Create an order here.
$ 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) {
// Get RawDB
db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB()
if err != nil {
return nil, status.Error(500, err.Error())
}
// Get sub transaction barrier object
barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx)
if err != nil {
return nil, status.Error(500, err.Error())
}
// Open sub transaction barrier
if err := barrier.CallWithDB(db, func(tx *sql.Tx) error {
// Query whether the user exists
_, err := l.svcCtx.UserRpc.UserInfo(l.ctx, &user.UserInfoRequest{
Id: in.Uid,
})
if err != nil {
return fmt.Errorf("user does not exist")
}
newOrder := model.Order{
Uid: in.Uid,
Pid: in.Pid,
Amount: in.Amount,
Status: 0,
}
// Create order
_, err = l.svcCtx.OrderModel.TxInsert(tx, &newOrder)
if err != nil {
return fmt.Errorf("Order creation failed")
}
return nil
}); err != nil {
return nil, status.Error(500, err.Error())
}
return &order.CreateResponse{}, nil
}Implement CreateRevert interface method
In this interface, we query the order just created by the user and change the status of the order to 9 (invalid status).
$ 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) {
// Get RawDB
db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB()
if err != nil {
return nil, status.Error(500, err.Error())
}
// Get sub transaction barrier object
barrier, err := dtmgrpc.BarrierFromGrpc(l.ctx)
if err != nil {
return nil, status.Error(500, err.Error())
}
// Open sub transaction barrier
if err := barrier.CallWithDB(db, func(tx *sql.Tx) error {
// Query whether the user exists
_, err := l.svcCtx.UserRpc.UserInfo(l.ctx, &user.UserInfoRequest{
Id: in.Uid,
})
if err != nil {
return fmt.Errorf("user does not exist")
}
// Query the latest order created by the user
resOrder, err := l.svcCtx.OrderModel.FindOneByUid(in.Uid)
if err != nil {
return fmt.Errorf("Order does not exist")
}
// Modify the order status 9, identify that the order has expired, and update the order
resOrder.Status = 9
err = l.svcCtx.OrderModel.TxUpdate(tx, resOrder)
if err != nil {
return fmt.Errorf("Order update failed")
}
return nil
}); err != nil {
return nil, status.Error(500, err.Error())
}
return &order.CreateResponse{}, nil
}10.2.6 modify order api service
We refer to the order rpc service Create and CreateRevert interface methods, and the product rpc service DecrStock and DecrStockRevert interface methods to make a distributed transaction operation in SAGA transaction mode in the order api service.
- Add pproduct rpc dependency configuration
$ 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- Add instantiation of pproduct rpc service configuration
$ 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
}- Register the dependency of the service context 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)),
}
}- Add dtm driver imported into gozero
$ vim mall/service/order/api/order.go
package main
import (
......
_ "github.com/dtm-labs/driver-gozero" // Add and import the 'dtm' driver of 'gozero'
)
var configFile = flag.String("f", "etc/order.yaml", "the config file")
func main() {
......
}- Modify the order api Create interface method
$ 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) {
// Get OrderRpc BuildTarget
orderRpcBusiServer, err := l.svcCtx.Config.OrderRpc.BuildTarget()
if err != nil {
return nil, status.Error(100, "Order creation exception")
}
// Get ProductRpc BuildTarget
productRpcBusiServer, err := l.svcCtx.Config.ProductRpc.BuildTarget()
if err != nil {
return nil, status.Error(100, "Order creation exception")
}
// etcd registration address of dtm service
var dtmServer = "etcd://etcd:2379/dtmservice"
// Create a gid
gid := dtmgrpc.MustGenGid(dtmServer)
// Create a transaction of saga protocol
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,
})
// Transaction commit
err = saga.Submit()
if err != nil {
return nil, status.Error(500, err.Error())
}
return &types.CreateResponse{}, nil
}Tip: sagagrpc The first parameter action of the add method is the method path accessed by the micro service grpc. This method path needs to be found in the following files respectively.
mall/service/order/rpc/order/order.pb.go
mall/service/product/rpc/product/product.pb.go
Search by keyword Invoke to find it.
10.3 test go zero + DTM
10.3.1 test the normal process of distributed transactions
- Use postman to call the / api/product/create interface to create a product with stock of 1.
- Use postman to call the / api/order/create interface to create an order with product ID pid 1.
- We can see that the inventory of products has changed from 1 to 0.
- When we look at the data in the sub transaction barrier table barrier, we can see that the operations of the two services have been completed.
10.3.2 test distributed transaction failure process 1
- Then, according to the above test results, the inventory with product ID 1 is already 0. Use postman to call the / api/order/create interface to create another order.
- Let's see that there is a piece of data with ID 2 and product ID 1 in the order data table, and its order data status is 9.
- When we look at the data in the sub transaction barrier table barrier, we can see that (gid = fqYS8CbYbK8GkL8SCuTRUF) the sub transaction barrier operation of the first service (branch_id = 01) is normal, and the sub transaction barrier operation of the second service (branch_id = 02) fails, requiring compensation. Therefore, the operation records of compensation occur in both services.
The operation flow of this distributed transaction
- First, the DTM service will call the order rpc Create interface to process the order creation.
- When the inventory of the product is insufficient through the RDEC service interface, the inventory of the product will be updated through the DTM interface.
- DTM service initiates the compensation mechanism and calls the order rpc CreateRevert interface to compensate the order.
- The DTM service initiates a compensation mechanism and calls the product rpc DecrStockRevert interface to compensate for product inventory updates. However, because the business processing is not successful within the sub transaction barrier of the product rpc DecrStock interface. Therefore, the business logic in the sub transaction barrier will not be executed in the DecrStockRevert interface.
10.3.3 test distributed transaction failure process 2
- We manually changed the inventory with product ID of 1 to 100 in the database, and then the artificial manufacturing exception failed outside the sub transaction barrier of the product rpc DecrStock interface method.
- Use postman to call the / api/order/create interface, and then create an order. The product ID pid is 1.
- Let's look at the order data table and product data table respectively. The order data table ID is 3, and its order data status is 9. For the product with ID 1 in the product data table, its inventory is still 100, and the data update time has also changed.
- Looking at the data in the sub transaction barrier table barrier, we can see that (gid = ZbjYHv2jNra7RMwyWjB5Lc) the sub transaction barrier operation of the first service (branch_id = 01) is normal, and the sub transaction barrier operation of the second service (branch_id = 02) is also normal. Because we artificially failed to create exceptions outside the sub transaction barrier of the product rpc DecrStock interface method, the two services had compensated operation records.
You can compare the differences between testing distributed transaction failure process 1 and testing distributed transaction failure process 2. Can you find and experience the strength of DTM's sub transaction barrier technology.
The sub transaction barrier will automatically identify whether the forward operation has been executed. The failed process 1 does not execute the business operation, so the compensated business operation will not be executed during compensation; Failed process 2 performs business operations, so the business operations of compensation will also be performed during compensation.
Project address
Welcome to go zero and star support us!
Wechat communication group
Focus on the "micro service practice" official account and click on the exchange group to get the community community's two-dimensional code.