Go-zero中分布式事务的实现(DTM分布式事务管理器,在一个APi中如何调用两个不同服务的rpc层,并保证两个不同服务之间的业务逻辑同时成功)

本文主要是介绍Go-zero中分布式事务的实现(DTM分布式事务管理器,在一个APi中如何调用两个不同服务的rpc层,并保证两个不同服务之间的业务逻辑同时成功)，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

涉及到的相关技术

1.DTM分布式事务管理器,解决跨数据库、跨服务、跨语言栈更新数据的一致性问题。

2.SAGA事务模式,SAGA事务模式是DTM中常用的一种模式,简单易上手.(当然还有其它更多的事务模式,这里采用的SAGA只不过是其中一种较为简单的方法)

3.Go-zero框架,ETCD服务注册...

更多内容移步至:go-zero 缩短从需求到上线的距离和介绍 | DTM开源项目文档

业务场景

如果是在单体架构的业务当中,是不需要用到分布式事务的.单体架构中,涉及到需要保证多个事务同时成功的场景,只需要创建一个全局的事务对象如:tx := db.Begin(),然后统一用这一个tx去管理接下来的业务逻辑即可.

不清楚在一个api中如何调用其它服务rpc的可以看看我的另一篇博客中的一种解决办法:

go-zero标准的项目结构,以及如何使用docker-compose部署道linux服务器上-CSDN博客

但是在go-zero框架的这种微服务中,比如说:我在一个用户服务的api中调用了用户服务rpc中注册的业务,并且同时还调用了标签服务的rpc层中的选择标签的业务. 那么,此时我就需要保证用户的注册和标签的选择这两个在不同服务下执行的业务逻辑同时成功.(总不能用户账号密码插入到的表中,但是突然断网了,导致标签没有选择上去吧,这个是不符合我的业务的).

DTM 环境搭建(Windows本地搭建)

!!!!!!!!!!!!!! 这个环境请注意,是需要在你本地去搭建的,至于为什么,我会在后面解释,最重要的先把环境搭建起来吧! 我采用的是docker-compose去搭建.(如果不了解windows电脑如何配置docker环境,可以移步:)

Windows11电脑是如何搭建docker环境的-CSDN博客

废话不多说,首先从搭建环境讲起.(我这里采用的是docker-compose搭建我需要的环境)

上图就是项目的结构

在dtm和etcd的目录下面各自新建一个Dockerfile文件,Dockerfile都不需要过多的配置,只需要用到最基础的镜像即可.在dtm的目录下还需要新建一个config.yml文件.

DTM下Dokcerfile以及config.yml的编写

FROM yedf/dtm:latestLABEL maintainer="zyf021026 <shichuxin6@163.com>"

# 指定要存储trans状态的存储驱动
# Store:### 默认存储驱动
#   Driver: 'boltdb'### redis 存储驱动
#   Driver: 'redis'
#   Host: 'localhost'
#   User: ''
#   Password: ''
#   Port: 6379### mysql 存储驱动
#   Driver: 'mysql'
#   Host: 'mysql'
#   User: 'root'
#   Password: '123456'
#   Port: 3306### postgres 存储驱动
#   Driver: 'postgres'
#   Host: 'localhost'
#   User: 'postgres'
#   Password: 'mysecretpassword'
#   Port: '5432'### 以下配置仅适用于 postgres/mysql 驱动
#   MaxOpenConns: 500
#   MaxIdleConns: 500
#   ConnMaxLifeTime: 5
#   TransGlobalTable: 'dtm.trans_global'
#   TransBranchOpTable: 'dtm.trans_branch_op'### 以下配置仅适用于 redis/boltdb 驱动
#   DataExpire: 604800 # Trans 过期时间
#   RedisPrefix: '{}'  # Redis 存储前缀MicroService:Driver: 'dtm-driver-gozero'           # 要处理注册/发现的驱动程序的名称Target: 'etcd://your-ip:2379/dtmservice' # 注册 dtm 服务的 etcd 地址EndPoint: 'your-ip:36790'# 以下配置的单位为'秒'
# TransCronInterval: 3
# TimeoutToFail: 35
# RetryInterval: 10# 日志等级
# LogLevel: 'info'

ETCD的Dockerfile文件编写

FROM bitnami/etcd:latestLABEL maintainer="zyf021026 <shichuxin6@163.com>"

使用docker-compose 构建镜像,启动容器


version: '3'networks:backend:driver: bridge######## 项目依赖的环境，启动项目之前要先启动此环境 #######
services:etcd:build:context: etcdenvironment:- TZ=Asia/Shanghai- ALLOW_NONE_AUTHENTICATION=yesports: # 设置端口映射- "2379:2379"networks:- backendrestart: alwaysdtm:build:context: ./dtmenvironment:- TZ=Asia/Shanghaientrypoint:- "/app/dtm/dtm"- "-c=/app/dtm/configs/config.yaml"privileged: truevolumes:- ./dtm/config.yml:/app/dtm/configs/config.yaml # 将 dtm 配置文件挂载到容器里ports:- "36789:36789"- "36790:36790"networks:- backendrestart: alwaysdepends_on:- etcd

在根目录下面执行docker-compose up -d 将需要的环境搭建起来

执行如下图中的命令:

新建子事务屏障的数据库(库名和表名请不要修改) 可以作为独立的一个数据库使用,没有必要把自己的项目数据库名称改为dtm_barrier

/*Navicat Premium Data TransferSource Server         : LinkSource Server Type    : MySQLSource Server Version : 50743Source Host           : 39.101.77.206:3306Source Schema         : dtm_barrierTarget Server Type    : MySQLTarget Server Version : 50743File Encoding         : 65001Date: 03/03/2024 13:57:48
*/SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS = 0;-- ----------------------------
-- Table structure for barrier
-- ----------------------------
DROP TABLE IF EXISTS `barrier`;
CREATE TABLE `barrier`  (`id` bigint(22) NOT NULL AUTO_INCREMENT,`trans_type` varchar(45) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT '',`gid` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT '',`branch_id` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT '',`op` varchar(45) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT '',`barrier_id` varchar(45) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT '',`reason` varchar(45) CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci NULL DEFAULT '' COMMENT 'the branch type who insert this record',`create_time` datetime NULL DEFAULT CURRENT_TIMESTAMP,`update_time` datetime NULL DEFAULT CURRENT_TIMESTAMP,PRIMARY KEY (`id`) USING BTREE,UNIQUE INDEX `gid`(`gid`, `branch_id`, `op`, `barrier_id`) USING BTREE,INDEX `create_time`(`create_time`) USING BTREE,INDEX `update_time`(`update_time`) USING BTREE
) ENGINE = InnoDB AUTO_INCREMENT = 1482 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_general_ci ROW_FORMAT = Dynamic;SET FOREIGN_KEY_CHECKS = 1;

SAGA事务模式的使用

简单说明一下,SAGA分布式事务模式,是没有办法携带返回值的,因此尽量此处要避免需要有返回值的业务场景.

直接用代码来展示SAGA事务模式的使用方法吧!

用户注册服务PRC的编写以及事务失败补偿机制的编写

这里不再演示proto文件是如何编写的

用户注册服务的rpc

func (l *UserCreateLogic) UserCreate(in *user.UserCreateRequest) (pd *user.UserCreateResponse, endErr error) {// 获取 RawDB// 注册db, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB()// 获取子事务屏障对象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 {// 加密密码pwd, _ := bcrypt.GetPwd(in.Password)// 插入用户数据_, err = tx.Exec("INSERT INTO users (id , created_at, updated_at, username, password, avatar, phone) VALUES (?,?, ?, ?, ?, ?, ?)", in.Id, time.Now(), time.Now(), in.Username, pwd, in.Avatar, in.Phone)//返回子事务执行失败if err != nil {return err}return nil})if err != nil {return nil, status.Error(codes.Aborted, dtmcli.ResultFailure) //如果失败,不再重试,直接回滚}return &user.UserCreateResponse{}, endErr
}

用户注册服务rpc的失败补偿 (如果注册服务的rpc失败,就会执行相应的补偿方法)

func (l *UserCreateRevertLoginLogic) UserCreateRevertLogin(in *user.UserCreateRequest) (pd *user.UserCreateResponse, err error) {fmt.Println("用户标签回滚开始--->")// 获取 RawDBdb, err := sqlx.NewMysql(l.svcCtx.Config.Mysql.DataSource).RawDB()// 获取子事务屏障对象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 {fmt.Println("注册事务走入了补偿")//删除插入的标签数据 和 用户数据_, err = tx.Exec("DELETE FROM tb_user_tag where user_id = ?", in.Id)_, err = tx.Exec("DELETE FROM users where id = ?", in.Id)//返回子事务执行失败if err != nil {return err}return nil})if err != nil {fmt.Println("failed---->", err)return nil, err}fmt.Println("删除成功")fmt.Println("用户标签回滚结束--->")return &user.UserCreateResponse{}, nil
}

标签服务Rpc的编写以及事务失败补偿机制的编写

标签服务的rpc

func (l *SignUserChooseTagLogic) SignUserChooseTag(in *tag.UserChooseTagRequest) (*tag.UserChooseTagRequest, 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)// 开启子事务屏障err = barrier.CallWithDB(db, func(tx *sql.Tx) (err error) {// 用户注册时选择标签var exists boolerr = tx.QueryRow("SELECT EXISTS(SELECT 1 FROM tb_user_tag WHERE tag_id = ? and user_id = ?)", in.TagId, in.UserId).Scan(&exists)if err != nil {return err}if exists {return fmt.Errorf("标签重复选择")}fmt.Println("开始插入标签")_, err = tx.Exec("INSERT INTO tb_user_tag (tb_user_tag.created_at , tb_user_tag.updated_at , tag_id, user_id) VALUES (?,?,?, ?)", time.Now(), time.Now(), in.TagId, in.UserId)if err != nil {return fmt.Errorf("标签选择失败")}return nil})if err != nil {return nil, status.Error(codes.Aborted, dtmcli.ResultFailure) //事务失败不再重试,直接回滚}return &tag.UserChooseTagRequest{}, nil
}

标签选择失败补偿的rpc

func (l *SignUserChooseTagRevertLogic) SignUserChooseTagRevert(in *tag.UserChooseTagRequest) (*tag.UserChooseTagRequest, error) {fmt.Println("用户标签SignUserChooseTagRevert--->开始")// 获取 RawDBdb, 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)// 开启子事务屏障err = barrier.CallWithDB(db, func(tx *sql.Tx) (err error) {fmt.Println("注册时选择标签进入了补偿")logc.Info(l.ctx)//删除记录_, err = tx.Exec("DELETE FROM tb_user_tag where tag_id = ? and user_id = ?", in.TagId, in.UserId)return err})if err != nil {return nil, err}fmt.Println("用户标签SignUserChooseTagRevert--->结束")return &tag.UserChooseTagRequest{}, nil
}

至此rpc层的业务逻辑全部编写完毕,但请一定要注意每一个rpc的返回值,一定要按照 如&tag.UserChooseTagRequest{}返回,不能简单的返回一个nil值.否则会导致事务一直无法提交

API层的编写


func (l *SignUpLogic) SignUp(req *types.UserCreateRequest) (resp *types.UserCreateResponse, err error) {//首先判断用户是否存在_, err = l.svcCtx.UserRpc.UserIsExists(l.ctx, &user.UserCreateRequest{Phone: req.Phone,})if err != nil {return nil, err}// 获取UserRpc 的BuildTargetuserRpcBuildServer, err := l.svcCtx.Config.UserRpc.BuildTarget()if err != nil {return nil, status.Error(100, "用户注册异常")}// 获取TagRpc 的BuildTargettagRpcBuildServer, err := l.svcCtx.Config.TagRpc.BuildTarget()if err != nil {return nil, status.Error(100, "标签选择异常")}empty := user.Empty{}//dtm服务的etcd注册地址var dtmServer = l.svcCtx.Config.Dtm//dtmServer := "etcd://etcd:2379/dtmservice"fmt.Println(dtmServer)// 创建一个gidgid := dtmgrpc.MustGenGid(dtmServer)//创建一个自增idif _, err := l.svcCtx.UserRpc.AddUserId(l.ctx, &empty); err != nil {return nil, fmt.Errorf("CREATE user id error:%v", err)}userID, _ := l.svcCtx.UserRpc.NextUserID(l.ctx, &empty)saga := dtmgrpc.NewSagaGrpc(dtmServer, gid).Add(tagRpcBuildServer+"/tag.TagSign/SignUserChooseTag", tagRpcBuildServer+"/tag.TagSign/SignUserChooseTagRevert", &tag.UserChooseTagRequest{UserId: userID.NextUserId,TagId:  req.StartTagId,}).Add(userRpcBuildServer+"/user.UserService/UserCreate", userRpcBuildServer+"/user.UserService/UserCreateRevertLogin", &user.UserCreateRequest{Username: req.Username,Password: req.Password,Avatar:   req.Avatar,Phone:    req.Phone,Id:       userID.NextUserId,})//事务提交if err := saga.Submit(); err != nil {//自增主键减少1if _, err := l.svcCtx.UserRpc.DecUserID(l.ctx, &empty); err != nil {logx.Error(err)}logx.Error(err)return nil, fmt.Errorf("saga submit error:%v", err)}return &types.UserCreateResponse{}, nil
}

上面代码的逻辑,相信如果各位接触到微服务,一定是可以理解的,由于saga的事务模式没有返回值,所以我通过redis生成一个自增id来使用,而不再采用mysql的自增主键id.