涉及到的相关技术
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:latest
LABEL 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:latest
LABEL maintainer="zyf021026 <shichuxin6@163.com>"使用docker-compose 构建镜像,启动容器
version: '3'
networks:
backend:
driver: bridge
######## 项目依赖的环境，启动项目之前要先启动此环境 #######
services:
etcd:
build:
context: etcd
environment:
- TZ=Asia/Shanghai
- ALLOW_NONE_AUTHENTICATION=yes
ports: # 设置端口映射
- "2379:2379"
networks:
- backend
restart: always
dtm:
build:
context: ./dtm
environment:
- TZ=Asia/Shanghai
entrypoint:
- "/app/dtm/dtm"
- "-c=/app/dtm/configs/config.yaml"
privileged: true
volumes:
- ./dtm/config.yml:/app/dtm/configs/config.yaml # 将 dtm 配置文件挂载到容器里
ports:
- "36789:36789"
- "36790:36790"
networks:
- backend
restart: always
depends_on:
- etcd
在根目录下面执行docker-compose up -d 将需要的环境搭建起来
执行如下图中的命令:
新建子事务屏障的数据库(库名和表名请不要修改) 可以作为独立的一个数据库使用,没有必要把自己的项目数据库名称改为dtm_barrier
/*
Navicat Premium Data Transfer
Source Server : Link
Source Server Type : MySQL
Source Server Version : 50743
Source Host : 39.101.77.206:3306
Source Schema : dtm_barrier
Target Server Type : MySQL
Target Server Version : 50743
File Encoding : 65001
Date: 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("用户标签回滚开始--->")
// 获取 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 {
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 bool
err = 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--->开始")
// 获取 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) {
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 的BuildTarget
userRpcBuildServer, err := l.svcCtx.Config.UserRpc.BuildTarget()
if err != nil {
return nil, status.Error(100, "用户注册异常")
}
// 获取TagRpc 的BuildTarget
tagRpcBuildServer, 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)
// 创建一个gid
gid := dtmgrpc.MustGenGid(dtmServer)
//创建一个自增id
if _, 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 {
//自增主键减少1
if _, 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.
上面代码中的地址可以在rpc生成的pb.go中找到
自己的理解
经历了长度一周多对分布式事务的研究,写一点自己的简单理解吧!(比较浅显)
saga事务模式需要自己写事务的补偿方法,子事务屏障内的事务执行失败之后,就会执行对应的事务补偿方法!即回滚事务.补偿方法内写的便是对这一次执行的插入,修改语句的相反操作.比如我增加某一条数据,补偿内就写上对删除的操作.
感觉和MySQL的Undo log 回滚日志很相似啊!Undo log日志会记录更新前的数据到日志中,是在一个事务下执行过程中,在还没有提交之前,如果发生意外,就可以通过这个日志回滚到事务执行之前的数据了.
![[备赛笔记]——5G大唐杯（5G考试等级考考试基础试题）](https://img-blog.csdnimg.cn/direct/135b53b5f5c443c28858992462ee4c98.gif)
