From 4918c54fc627cf634b70902f6d7224677d565dbc Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Wed, 11 Dec 2024 16:10:56 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E6=8E=A5=E5=8F=A3=E5=8F=8A?= =?UTF-8?q?=EF=BC=8C=E8=8A=82=E7=82=B9=E8=8E=B7=E5=8F=96=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=8F=8A=E7=B3=BB=E7=BB=9F=E5=AE=9A=E6=97=B6=E4=BB=BB=E5=8A=A1?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dao/user.go | 27 +++++++++++++++++++++++++++ main.go | 41 +++++++++++++++++++++++++++++++++++++++-- proto/conf.go | 31 +++++++++++++++++-------------- proto/user_req.go | 39 ++++++++++++++++++++++++++++++++++++++- service/userService.go | 35 +++++++++++++++++++++++++++++++++++ worker/req.go | 38 ++++++++++++++++++++++++++++++++++++++ 6 files changed, 194 insertions(+), 17 deletions(-) diff --git a/dao/user.go b/dao/user.go index 1f8d702..c474040 100644 --- a/dao/user.go +++ b/dao/user.go @@ -101,3 +101,30 @@ func UpdateUserByID2(id int, req proto.UpdateUserInfoReq) { func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) { DB.Model(&User{}).Where("id = ?", id).Updates(User{Name: req.Username, Age: req.Age, Avatar: req.Avatar, Gender: req.Gender}) } + +// 用户数据同步-添加 +func AddUserSync(req proto.UserAddOrUpdate) uint { + res := DB.Exec("insert into users (id, created_at, updated_at, deleted_at, name, age, email, password,gender,role,redis,run,upload,video_func,device_func,cid_func,avatar,create_time,update_time) values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", req.ID, req.CreatedAt, req.UpdatedAt, req.DeletedAt, req.Name, req.Age, req.Email, req.Password, req.Gender, req.Role, req.Redis, req.Run, req.Upload, req.VideoFunc, req.DeviceFunc, req.CIDFunc, req.Avatar, req.CreateTime, req.UpdateTime) + if res.Error != nil { + return 0 + } + return req.ID +} + +// 用户数据同步-更新 +func UpdateUserSync(req proto.UserAddOrUpdate) uint { + res := DB.Exec("update users set created_at=?, updated_at=?, deleted_at=?, name=?, age=?, email=?, password=?,gender=?,role=?,redis=?,run=?,upload=?,video_func=?,device_func=?,cid_func=?,avatar=?,create_time=?,update_time=? where id=?", req.CreatedAt, req.UpdatedAt, req.DeletedAt, req.Name, req.Age, req.Email, req.Password, req.Gender, req.Role, req.Redis, req.Run, req.Upload, req.VideoFunc, req.DeviceFunc, req.CIDFunc, req.Avatar, req.CreateTime, req.UpdateTime, req.ID) + if res.Error != nil { + return 0 + } + return req.ID +} + +// 用户数据同步-删除 +func DeleteUserSync(req proto.UserDelID) uint { + res := DB.Delete(&User{}, req.ID) + if res.Error != nil { + return 0 + } + return req.ID +} diff --git a/main.go b/main.go index 186785d..a2114e3 100644 --- a/main.go +++ b/main.go @@ -13,6 +13,7 @@ import ( "videoplayer/dao" "videoplayer/handler" "videoplayer/proto" + "videoplayer/service" "videoplayer/worker" ) @@ -222,8 +223,25 @@ func ReadConfigToSetSystem() { logClean.Every = 86400 cron_infos = append(cron_infos, logClean) } - } + if proto.Config.SERVER_USER_TYPE == "slave" && proto.Config.USER_SYNC_TIME > 0 { + var is_exist bool + for _, v := range cron_infos { + if v.Type == 2 { + is_exist = true + break + } + } + if !is_exist { + var userSync proto.CronInfo + userSync.Type = 2 + userSync.Info = "user" + userSync.Curr = proto.Config.USER_SYNC_TIME + userSync.Every = proto.Config.USER_SYNC_TIME + cron_infos = append(cron_infos, userSync) + } + } + } else { if proto.Config.LOG_SAVE_DAYS > 0 { var logClean proto.CronInfo @@ -233,6 +251,14 @@ func ReadConfigToSetSystem() { logClean.Every = 86400 cron_infos = append(cron_infos, logClean) } + if proto.Config.SERVER_USER_TYPE == "slave" && proto.Config.USER_SYNC_TIME > 0 { + var userSync proto.CronInfo + userSync.Type = 2 + userSync.Info = "user" + userSync.Curr = proto.Config.USER_SYNC_TIME + userSync.Every = proto.Config.USER_SYNC_TIME + cron_infos = append(cron_infos, userSync) + } } //存入redis json_data, err := json.Marshal(cron_infos) @@ -266,6 +292,18 @@ func RunGeneralCron() { v.Curr -= 10 } } + //2 从服务器同步数据 + if v.Type == 2 { + if v.Curr <= 0 { + //执行从服务器同步数据 + if proto.Config.SERVER_USER_TYPE == "slave" && v.Info == "user" { + go service.UserSyncDataFromMaster() + } + v.Curr = v.Every + } else { + v.Curr -= 10 + } + } } //存入redis json_data, err := json.Marshal(cron_infos) @@ -274,7 +312,6 @@ func RunGeneralCron() { } else { worker.SetRedis(key, string(json_data)) } - } } diff --git a/proto/conf.go b/proto/conf.go index d014382..359020f 100644 --- a/proto/conf.go +++ b/proto/conf.go @@ -63,20 +63,23 @@ type User struct { } type ConfigStruct struct { - DB int `json:"db"` // 0: mysql, 1: pg - MYSQL_DSN string `json:"mysql_dsn"` - PG_DSN string `json:"pg_dsn"` - REDIS_ADDR string `json:"redis_addr"` - TOKEN_USE_REDIS bool `json:"token_use_redis"` - REDIS_User_PW bool `json:"redis_user_pw"` // 是否使用密码 - REDIS_PASSWORD string `json:"redis_password"` - REDIS_DB int `json:"redis_db"` - TOKEN_SECRET string `json:"token_secret"` - CID_BASE_DIR string `json:"cid_base_dir"` - FILE_BASE_DIR string `json:"file_base_dir"` - MONITOR bool `json:"monitor"` // 状态监控及邮件通知 - SERVER_PORT string `json:"server_port"` // 服务端口 - LOG_SAVE_DAYS int `json:"log_save_days"` // 日志保存天数,-1表示不保存，0表示永久保存 + DB int `json:"db"` // 0: mysql, 1: pg + MYSQL_DSN string `json:"mysql_dsn"` + PG_DSN string `json:"pg_dsn"` + REDIS_ADDR string `json:"redis_addr"` + TOKEN_USE_REDIS bool `json:"token_use_redis"` + REDIS_User_PW bool `json:"redis_user_pw"` // 是否使用密码 + REDIS_PASSWORD string `json:"redis_password"` + REDIS_DB int `json:"redis_db"` + TOKEN_SECRET string `json:"token_secret"` + CID_BASE_DIR string `json:"cid_base_dir"` + FILE_BASE_DIR string `json:"file_base_dir"` + MONITOR bool `json:"monitor"` // 状态监控及邮件通知 + SERVER_PORT string `json:"server_port"` // 服务端口 + LOG_SAVE_DAYS int `json:"log_save_days"` // 日志保存天数,-1表示不保存，0表示永久保存 + SERVER_USER_TYPE string `json:"user_type"` // 服务器用户类型，master: 主服务器，slave: 从服务器，从服务器会定时同步数据 + MASTER_SERVER_DOMAIN string `json:"master_server_domain"` // 主服务器域名 + USER_SYNC_TIME int `json:"user_sync_time"` // 用户数据同步时间，单位秒 } // 读取配置文件 diff --git a/proto/user_req.go b/proto/user_req.go index a42f24b..eea57ec 100644 --- a/proto/user_req.go +++ b/proto/user_req.go @@ -1,5 +1,9 @@ package proto +import ( + "time" +) + type UpdateUserInfoReq struct { ID int `json:"id" form:"id"` //用户id Username string `json:"name" form:"name"` //用户名 @@ -23,8 +27,41 @@ type CIDRUN struct { // 用于执行函数，方法 type CronInfo struct { - Type int `json:"type" form:"type"` //类型编码,1日志清理（且只会有一个），其他待定 + Type int `json:"type" form:"type"` //类型编码,1日志清理（且只会有一个），其他待定，2从服务器同步数据 Info string `json:"info" form:"info"` //信息 Curr int `json:"curr" form:"curr"` //当前剩余时间，每次执行减10s小于等于0则执行 Every int `json:"every" form:"every"` //每隔多少秒执行一次,小于等于0表示不执行，时间粒度为10s } + +// 用户数据同步 +type UserSync struct { + Update []UserAddOrUpdate `json:"update" form:"update"` //更新用户 + Add []UserAddOrUpdate `json:"add" form:"add"` //添加用户 + Delete []UserDelID `json:"delete" form:"delete"` //删除用户 +} + +type UserDelID struct { + ID uint `json:"ID" form:"ID"` //用户id +} + +type UserAddOrUpdate struct { + ID uint `json:"ID" form:"ID"` //用户id + CreatedAt time.Time `json:"CreatedAt" form:"CreatedAt"` //创建时间 + UpdatedAt time.Time `json:"UpdatedAt" form:"UpdatedAt"` //更新时间 + DeletedAt time.Time `json:"DeletedAt" form:"DeletedAt"` //删除时间 + Name string `json:"Name" form:"Name"` //用户名 + Age int `json:"Age" form:"Age"` //年龄 + Email string `json:"Email" form:"Email"` //邮箱 + Password string `json:"Password" form:"Password"` //密码 + Gender string `json:"Gender" form:"Gender"` //性别 + Role string `json:"Role" form:"Role"` //角色 + Redis bool `json:"Redis" form:"Redis"` //是否刷新redis + Run bool `json:"Run" form:"Run"` //是否运行 + Upload bool `json:"Upload" form:"Upload"` //是否上传头像 + VideoFunc bool `json:"VideoFunc" form:"VideoFunc"` //视频功能 + DeviceFunc bool `json:"DeviceFunc" form:"DeviceFunc"` + CIDFunc bool `json:"CIDFunc" form:"CIDFunc"` + Avatar string `json:"Avatar" form:"Avatar"` //头像 + CreateTime string `json:"CreateTime" form:"CreateTime"` + UpdateTime string `json:"UpdateTime" form:"UpdateTime"` +} diff --git a/service/userService.go b/service/userService.go index b395b94..eae4339 100644 --- a/service/userService.go +++ b/service/userService.go @@ -4,6 +4,7 @@ import ( "regexp" "videoplayer/dao" "videoplayer/proto" + "videoplayer/worker" ) func CreateUser(name, password, email, gender string, age int) uint { @@ -55,3 +56,37 @@ func UpdateUser(user_id int, req proto.UpdateUserInfoReq) (int, error) { return 0, nil } } + +func UserSyncDataFromMaster() { + //从接口获取数据 + url := "http://" + proto.Config.MASTER_SERVER_DOMAIN + "/user/sync" + tokens := worker.GetRedisSetMembers("super_permission_tokens") + user_sync_data := worker.SyncDataFromMasterReq(url, tokens[0]) + add_users := user_sync_data.Add + update_users := user_sync_data.Update + delete_users := user_sync_data.Delete + //未成功操作的id + var fail_ids []uint + + //添加用户 + for _, v := range add_users { + res := dao.AddUserSync() + if res == 0 { + fail_ids = append(fail_ids, v.ID) + } + } + //更新用户 + for _, v := range update_users { + res := dao.UpdateUserSync(v) + if res == 0 { + fail_ids = append(fail_ids, v.ID) + } + } + //删除用户 + for _, v := range delete_users { + res := dao.DeleteUserSync(v) + if res == 0 { + fail_ids = append(fail_ids, v.ID) + } + } +} diff --git a/worker/req.go b/worker/req.go index e271110..0b42692 100644 --- a/worker/req.go +++ b/worker/req.go @@ -3,10 +3,12 @@ package worker import ( "bytes" "encoding/json" + "fmt" "io" "io/ioutil" "net/http" "strings" + "videoplayer/proto" ) var client *http.Client @@ -80,3 +82,39 @@ func GenerateCompletion(url, prompt string, model string) (map[string]interface{ return result, nil } + +// 获取同步数据通用方法 +func SyncDataFromMasterReq(url string, token string) proto.UserSync { + //从接口获取数据 + req, err := http.NewRequest("POST", url, nil) + if err != nil { + return proto.UserSync{} + } + req.Header.Set("token", token) + client := &http.Client{} + //获取数据 + resp, err := client.Do(req) + if err != nil { + return proto.UserSync{} + } + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + if err != nil { + return proto.UserSync{} + } + var result map[string]interface{} + err = json.Unmarshal(body, &result) + if err != nil { + return proto.UserSync{} + } + fmt.Println("SyncDataFromMasterReq result:", result) + if result["code"].(float64) != 0 { + return proto.UserSync{} + } + var userSync proto.UserSync + err = json.Unmarshal([]byte(result["data"].(string)), &userSync) + if err != nil { + return proto.UserSync{} + } + return userSync +} From 29e8151420f7655cc7a112d8f013160d4e72425d Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Wed, 11 Dec 2024 17:43:44 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E7=94=A8=E6=88=B7=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=90=8C=E6=AD=A5=E9=83=A8=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dao/user.go | 7 +++++++ handler/user.go | 42 ++++++++++++++++++++++++++++++++++++++++++ proto/user_req.go | 8 ++++++++ service/userService.go | 26 ++++++++++++++++++++++++++ 4 files changed, 83 insertions(+) diff --git a/dao/user.go b/dao/user.go index c474040..c6236b2 100644 --- a/dao/user.go +++ b/dao/user.go @@ -128,3 +128,10 @@ func DeleteUserSync(req proto.UserDelID) uint { } return req.ID } + +// 获取所有用户 +func GetAllUser() []proto.User { + var users []proto.User + DB.Find(&users) + return users +} diff --git a/handler/user.go b/handler/user.go index 11023d0..03a44e0 100644 --- a/handler/user.go +++ b/handler/user.go @@ -26,6 +26,7 @@ func SetUpUserGroup(router *gin.Engine) { userGroup.POST("/search", SearchHandler) userGroup.POST("/info", GetUserInfo) userGroup.POST("/update", UpdateUserInfo) + userGroup.POST("/sync", GetSyncUserInfo) } type RLReq struct { @@ -350,3 +351,44 @@ func registerHandler(c *gin.Context) { c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": data}) return } + +type SyncUserReq struct { + token string `json:"token" form:"token"` + types int `json:"type" form:"type"` // 1为全量同步 2为增量同步 +} + +func GetSyncUserInfo(c *gin.Context) { + var req_data SyncUserReq + if err := c.ShouldBind(&req_data); err == nil { + if req_data.token == "" { + c.JSON(200, gin.H{"code": proto.ParameterError, "message": "error", "data": "token is empty"}) + return + } else { + if worker.IsContainSet("super_permission_tokens", req_data.token) { + if proto.Config.SERVER_USER_TYPE == "master" { + if req_data.types == 1 { + add_users := dao.GetAllUser() + resp := proto.UserSyncResp{} + resp.Add = add_users + c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": resp}) + } else if req_data.types == 2 { + res := service.GetUserSyncData() + c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res}) + } else { + c.JSON(200, gin.H{"code": proto.ParameterError, "message": "type is error", "data": proto.UserSyncResp{}}) + } + + } else { + c.JSON(200, gin.H{"code": proto.NoPermission, "message": "no permission,server is not master", "data": proto.UserSync{}}) + } + } else { + c.JSON(200, gin.H{"code": proto.NoPermission, "message": "error", "data": "no permission"}) + return + } + } + + } else { + c.JSON(200, gin.H{"error": err.Error(), "code": proto.ParameterError, "message": "error"}) + return + } +} diff --git a/proto/user_req.go b/proto/user_req.go index eea57ec..1aa18e8 100644 --- a/proto/user_req.go +++ b/proto/user_req.go @@ -2,6 +2,7 @@ package proto import ( "time" + "videoplayer/dao" ) type UpdateUserInfoReq struct { @@ -33,6 +34,13 @@ type CronInfo struct { Every int `json:"every" form:"every"` //每隔多少秒执行一次,小于等于0表示不执行，时间粒度为10s } +// 用户数据同步 +type UserSyncResp struct { + Update []dao.User `json:"update" form:"update"` //更新用户 + Add []dao.User `json:"add" form:"add"` //添加用户 + Delete []UserDelID `json:"delete" form:"delete"` //删除用户 +} + // 用户数据同步 type UserSync struct { Update []UserAddOrUpdate `json:"update" form:"update"` //更新用户 diff --git a/service/userService.go b/service/userService.go index eae4339..ba5acbe 100644 --- a/service/userService.go +++ b/service/userService.go @@ -2,6 +2,7 @@ package service import ( "regexp" + "strconv" "videoplayer/dao" "videoplayer/proto" "videoplayer/worker" @@ -90,3 +91,28 @@ func UserSyncDataFromMaster() { } } } + +// 同步数据到主服务器-增删改数据 +func GetUserSyncData() proto.UserSyncResp { + add_user_ids := worker.GetRedisSetMembers("sync_add_user_ids") + update_user_ids := worker.GetRedisSetMembers("sync_update_user_ids") + delete_user_ids := worker.GetRedisSetMembers("sync_delete_user_ids") + add_users := []dao.User{} + update_users := []dao.User{} + delete_users := []proto.UserDelID{} + for _, v := range add_user_ids { + id, _ := strconv.Atoi(v) + user := dao.FindUserByUserID(id) + add_users = append(add_users, user) + } + for _, v := range update_user_ids { + id, _ := strconv.Atoi(v) + user := dao.FindUserByUserID(id) + update_users = append(update_users, user) + } + for _, v := range delete_user_ids { + id, _ := strconv.Atoi(v) + delete_users = append(delete_users, proto.UserDelID{ID: uint(id)}) + } + return proto.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users} +} From 8796ef784b49da26e445f64020df6e00ee1eb5b5 Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Thu, 12 Dec 2024 21:08:53 +0800 Subject: [PATCH 3/6] =?UTF-8?q?=E5=AE=8C=E5=96=84=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=EF=BC=8C=E5=BE=85=E5=AE=8C?= =?UTF-8?q?=E5=96=84=E5=90=8C=E6=AD=A5=E7=A1=AE=E8=AE=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- dao/user.go | 17 +++++++--- handler/user.go | 38 +++++++++++++++++----- service/userService.go | 73 +++++++++++++++++++++++++++++++++++++----- 3 files changed, 107 insertions(+), 21 deletions(-) diff --git a/dao/user.go b/dao/user.go index c6236b2..43aec3a 100644 --- a/dao/user.go +++ b/dao/user.go @@ -35,7 +35,10 @@ func CreateUser(name, password, email, gender string, age int) uint { } func DeleteUserByID(id int) int { - DB.Delete(&User{}, id) + res := DB.Delete(&User{}, id) + if res.Error != nil { + return 0 + } return id } @@ -98,8 +101,12 @@ func UpdateUserByID2(id int, req proto.UpdateUserInfoReq) { } // 用户修改自己的信息 -func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) { - DB.Model(&User{}).Where("id = ?", id).Updates(User{Name: req.Username, Age: req.Age, Avatar: req.Avatar, Gender: req.Gender}) +func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) error { + res := DB.Model(&User{}).Where("id = ?", id).Updates(User{Name: req.Username, Age: req.Age, Avatar: req.Avatar, Gender: req.Gender}) + if res.Error != nil { + return res.Error + } + return nil } // 用户数据同步-添加 @@ -130,8 +137,8 @@ func DeleteUserSync(req proto.UserDelID) uint { } // 获取所有用户 -func GetAllUser() []proto.User { - var users []proto.User +func GetAllUser() []User { + var users []User DB.Find(&users) return users } diff --git a/handler/user.go b/handler/user.go index 03a44e0..26cfd5c 100644 --- a/handler/user.go +++ b/handler/user.go @@ -27,6 +27,7 @@ func SetUpUserGroup(router *gin.Engine) { userGroup.POST("/info", GetUserInfo) userGroup.POST("/update", UpdateUserInfo) userGroup.POST("/sync", GetSyncUserInfo) + userGroup.POST("/delete", DeleteUser) } type RLReq struct { @@ -78,6 +79,23 @@ func GetUserInfo(c *gin.Context) { } } +func DeleteUser(c *gin.Context) { + var req GetUserInfoReq + id, _ := c.Get("id") + user_id := int(id.(float64)) + if err := c.ShouldBind(&req); err == nil { + res := service.DeleteUserService(req.ID, user_id) + if res != 0 { + c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res}) + } else { + c.JSON(200, gin.H{"code": proto.OperationFailed, "message": "failed", "data": res}) + } + } else { + c.JSON(200, gin.H{"code": proto.ParameterError, "message": err, "data": "2"}) + return + } +} + func UpdateUserInfo(c *gin.Context) { var req_data proto.UpdateUserInfoReq id, _ := c.Get("id") @@ -353,31 +371,35 @@ func registerHandler(c *gin.Context) { } type SyncUserReq struct { - token string `json:"token" form:"token"` - types int `json:"type" form:"type"` // 1为全量同步 2为增量同步 + Token string `json:"token" form:"token"` + Types int `json:"type" form:"type"` // 1为全量同步 2为增量同步 + Device string `json:"device" form:"device"` } func GetSyncUserInfo(c *gin.Context) { var req_data SyncUserReq if err := c.ShouldBind(&req_data); err == nil { - if req_data.token == "" { + if req_data.Token == "" { c.JSON(200, gin.H{"code": proto.ParameterError, "message": "error", "data": "token is empty"}) return } else { - if worker.IsContainSet("super_permission_tokens", req_data.token) { + if worker.IsContainSet("super_permission_tokens", req_data.Token) { if proto.Config.SERVER_USER_TYPE == "master" { - if req_data.types == 1 { + if req_data.Types == 1 { //1为全量同步 add_users := dao.GetAllUser() resp := proto.UserSyncResp{} resp.Add = add_users c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": resp}) - } else if req_data.types == 2 { - res := service.GetUserSyncData() + } else if req_data.Types == 2 { //2为增量同步 + if req_data.Device == "" || worker.IsContainSet("sync_devices_ids", req_data.Device) == false { + c.JSON(200, gin.H{"code": proto.ParameterError, "message": "error", "data": "device is empty or not exist"}) + return + } + res := service.GetUserSyncData(req_data.Device) c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res}) } else { c.JSON(200, gin.H{"code": proto.ParameterError, "message": "type is error", "data": proto.UserSyncResp{}}) } - } else { c.JSON(200, gin.H{"code": proto.NoPermission, "message": "no permission,server is not master", "data": proto.UserSync{}}) } diff --git a/service/userService.go b/service/userService.go index ba5acbe..10da872 100644 --- a/service/userService.go +++ b/service/userService.go @@ -1,6 +1,7 @@ package service import ( + "errors" "regexp" "strconv" "videoplayer/dao" @@ -9,7 +10,15 @@ import ( ) func CreateUser(name, password, email, gender string, age int) uint { - return dao.CreateUser(name, password, email, gender, age) + id := dao.CreateUser(name, password, email, gender, age) + if id != 0 { + //添加用户信息到同步列表 + err := setSyncUserDataSet("add", int(id)) + if err != nil { + return id + } + } + return id } func GetUser(name, email, password string) dao.User { @@ -48,8 +57,15 @@ func UpdateUser(user_id int, req proto.UpdateUserInfoReq) (int, error) { cur_user := dao.FindUserByID2(user_id) //fmt.Println("cur_user:", cur_user, "req:", req) if user_id == req.ID && cur_user.Role != "admin" { - dao.UpdateUserByID3(user_id, req) //用户修改自己的信息，不能修改权限信息 - return user_id, nil + err := dao.UpdateUserByID3(user_id, req) //用户修改自己的信息，不能修改权限信息 + //添加修改用户信息到同步列表 + if err == nil { + err2 := setSyncUserDataSet("update", user_id) + if err2 != nil { + return user_id, nil + } + } + return user_id, err } else if cur_user.Role == "admin" { dao.UpdateUserByID2(req.ID, req) return req.ID, nil @@ -58,6 +74,25 @@ func UpdateUser(user_id int, req proto.UpdateUserInfoReq) (int, error) { } } +func DeleteUserService(id, user_id int) int { + res := 0 + if user_id == id { + res = dao.DeleteUserByID(id) + } else { + user := dao.FindUserByID2(user_id) + if user.Role == "admin" { + res = dao.DeleteUserByID(id) + } + } + if res != 0 { + //添加删除用户信息到同步列表 + err := setSyncUserDataSet("delete", id) + if err != nil { + return res + } + } + return res +} func UserSyncDataFromMaster() { //从接口获取数据 url := "http://" + proto.Config.MASTER_SERVER_DOMAIN + "/user/sync" @@ -71,7 +106,7 @@ func UserSyncDataFromMaster() { //添加用户 for _, v := range add_users { - res := dao.AddUserSync() + res := dao.AddUserSync(v) if res == 0 { fail_ids = append(fail_ids, v.ID) } @@ -93,10 +128,11 @@ func UserSyncDataFromMaster() { } // 同步数据到主服务器-增删改数据 -func GetUserSyncData() proto.UserSyncResp { - add_user_ids := worker.GetRedisSetMembers("sync_add_user_ids") - update_user_ids := worker.GetRedisSetMembers("sync_update_user_ids") - delete_user_ids := worker.GetRedisSetMembers("sync_delete_user_ids") +func GetUserSyncData(device string) proto.UserSyncResp { + key := device + "_sync_user_ids" + add_user_ids := worker.GetRedisSetMembers(key + "_add") + update_user_ids := worker.GetRedisSetMembers(key + "_update") + delete_user_ids := worker.GetRedisSetMembers(key + "_delete") add_users := []dao.User{} update_users := []dao.User{} delete_users := []proto.UserDelID{} @@ -116,3 +152,24 @@ func GetUserSyncData() proto.UserSyncResp { } return proto.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users} } + +func setSyncUserDataSet(t string, id int) error { + devices := worker.GetRedisSetMembers("sync_devices_ids") //主服务器查看从服务器的设备列表 + var err error + for _, device := range devices { + key := device + "_sync_user_ids" + if t == "add" { + key_ := key + "_add" + worker.SetRedisSetAdd(key_, strconv.Itoa(id)) + } else if t == "update" { + key_ := key + "_update" + worker.SetRedisSetAdd(key_, strconv.Itoa(id)) + } else if t == "delete" { + key_ := key + "_delete" + worker.SetRedisSetAdd(key_, strconv.Itoa(id)) + } else { + err = errors.New("error") + } + } + return err +} From 3571772bb1d04d5137b4f1bae0884fd25b8080b8 Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Fri, 13 Dec 2024 20:21:05 +0800 Subject: [PATCH 4/6] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E7=A1=AE=E8=AE=A4=E9=83=A8?= =?UTF-8?q?=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- handler/user.go | 14 +++++++--- proto/user_req.go | 11 ++++++++ service/userService.go | 61 ++++++++++++++++++++++++++++++++++++++++++ worker/redis.go | 50 ++++++++++++++++++++++++++++++++++ 4 files changed, 133 insertions(+), 3 deletions(-) diff --git a/handler/user.go b/handler/user.go index 26cfd5c..eb06568 100644 --- a/handler/user.go +++ b/handler/user.go @@ -371,9 +371,10 @@ func registerHandler(c *gin.Context) { } type SyncUserReq struct { - Token string `json:"token" form:"token"` - Types int `json:"type" form:"type"` // 1为全量同步 2为增量同步 - Device string `json:"device" form:"device"` + Token string `json:"token" form:"token"` + Types int `json:"type" form:"type"` // 1为全量同步 2为增量同步 + Device string `json:"device" form:"device"` + Confirm proto.UserSyncConfirm `json:"confirm" form:"confirm"` } func GetSyncUserInfo(c *gin.Context) { @@ -397,6 +398,13 @@ func GetSyncUserInfo(c *gin.Context) { } res := service.GetUserSyncData(req_data.Device) c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": res}) + } else if req_data.Types == 3 { //3为确认同步数据 + res := service.ConfirmSyncUserData(req_data.Device, req_data.Confirm) + if res == nil { + c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": "success"}) + } else { + c.JSON(200, gin.H{"code": proto.OperationFailed, "message": "failed", "data": "failed"}) + } } else { c.JSON(200, gin.H{"code": proto.ParameterError, "message": "type is error", "data": proto.UserSyncResp{}}) } diff --git a/proto/user_req.go b/proto/user_req.go index 1aa18e8..6711302 100644 --- a/proto/user_req.go +++ b/proto/user_req.go @@ -48,6 +48,17 @@ type UserSync struct { Delete []UserDelID `json:"delete" form:"delete"` //删除用户 } +// 用户数据同步确认 +type UserSyncConfirm struct { + Add []UserConfirmID `json:"add" form:"add"` //添加用户 + Update []UserConfirmID `json:"update" form:"update"` //更新用户 + Delete []UserConfirmID `json:"delete" form:"delete"` //删除用户 +} + +type UserConfirmID struct { + ID uint `json:"id" form:"id"` //用户id +} + type UserDelID struct { ID uint `json:"ID" form:"ID"` //用户id } diff --git a/service/userService.go b/service/userService.go index 10da872..0a7985e 100644 --- a/service/userService.go +++ b/service/userService.go @@ -4,6 +4,7 @@ import ( "errors" "regexp" "strconv" + "time" "videoplayer/dao" "videoplayer/proto" "videoplayer/worker" @@ -141,15 +142,27 @@ func GetUserSyncData(device string) proto.UserSyncResp { user := dao.FindUserByUserID(id) add_users = append(add_users, user) } + for _, v := range update_user_ids { id, _ := strconv.Atoi(v) user := dao.FindUserByUserID(id) update_users = append(update_users, user) } + for _, v := range delete_user_ids { id, _ := strconv.Atoi(v) delete_users = append(delete_users, proto.UserDelID{ID: uint(id)}) } + //将id存入暂存集合，清空原集合 + add_temp_key := device + "_sync_user_ids_add_confirm_temp" + update_temp_key := device + "_sync_user_ids_update_confirm_temp" + delete_temp_key := device + "_sync_user_ids_delete_confirm_temp" + worker.SetRedisSetUnionAndStore(add_temp_key, key+"_add") + worker.SetRedisSetClear(key + "_add") + worker.SetRedisSetUnionAndStore(update_temp_key, key+"_update") + worker.SetRedisSetClear(key + "_update") + worker.SetRedisSetUnionAndStore(delete_temp_key, key+"_delete") + worker.SetRedisSetClear(key + "_delete") return proto.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users} } @@ -173,3 +186,51 @@ func setSyncUserDataSet(t string, id int) error { } return err } + +// 确认同步数据 +func ConfirmSyncUserData(device string, data proto.UserSyncConfirm) error { + + var ids_add []string + var err error + for _, v := range data.Add { + ids_add = append(ids_add, strconv.Itoa(int(v.ID))) + } + add_key := device + "_sync_user_ids_add_confirm" + isSuccess := worker.SetRedisSetAddBatchWithExpire(add_key, ids_add, time.Second*30) + if !isSuccess { + err = errors.New("set confirm error") + } + var ids_update []string + for _, v := range data.Update { + ids_update = append(ids_update, strconv.Itoa(int(v.ID))) + } + update_key := device + "_sync_user_ids_update_confirm" + isSuccess = worker.SetRedisSetAddBatchWithExpire(update_key, ids_update, time.Second*30) + if !isSuccess { + err = errors.New("set confirm error") + } + + var ids_delete []string + for _, v := range data.Delete { + ids_delete = append(ids_delete, strconv.Itoa(int(v.ID))) + } + del_key := device + "_sync_user_ids_delete_confirm" + isSuccess = worker.SetRedisSetAddBatchWithExpire(del_key, ids_delete, time.Second*30) + if !isSuccess { + err = errors.New("set confirm error") + } + + //待确认集合暂存 + ids_add_confirm_temp := device + "_sync_user_ids_add_confirm_temp" + ids_update_confirm_temp := device + "_sync_user_ids_update_confirm_temp" + ids_delete_confirm_temp := device + "_sync_user_ids_delete_confirm_temp" + + //取差集 + add_diff := worker.SetRedisSetDiffAndStore(ids_add_confirm_temp, add_key) + update_diff := worker.SetRedisSetDiffAndStore(ids_update_confirm_temp, update_key) + delete_diff := worker.SetRedisSetDiffAndStore(ids_delete_confirm_temp, del_key) + if add_diff != true || update_diff != true || delete_diff != true { + err = errors.New("confirm error") + } + return err +} diff --git a/worker/redis.go b/worker/redis.go index a9957bb..84bc598 100644 --- a/worker/redis.go +++ b/worker/redis.go @@ -292,6 +292,23 @@ func SetRedisSetAdd(key string, value string) bool { return true } +// 批量添加元素 +func SetRedisSetAddBatchWithExpire(key string, values []string, expire time.Duration) bool { + ctx := context.Background() + err := RedisClient.SAdd(ctx, key, values).Err() + if err != nil { + fmt.Println("Error setting key: %v", err) + return false + } + err = RedisClient.Expire(ctx, key, expire).Err() + if err != nil { + fmt.Println("Error setting key: %v", err) + return false + } + return true + +} + // 设置set,添加元素 func SetRedisSetAddWithExpire(key string, value string, expire time.Duration) bool { ctx := context.Background() @@ -399,3 +416,36 @@ func Subscribe(channel string) []string { } return messages } + +// redis两个set求差集存入第一个set +func SetRedisSetDiffAndStore(key1 string, key2 string) bool { + ctx := context.Background() + err := RedisClient.SDiffStore(ctx, key1, key1, key2).Err() //将key1和key2的差集存入key1 + if err != nil { + fmt.Println("Error setting key: %v", err) + return false + } + return true +} + +// redis将第二个set存入第一个set +func SetRedisSetUnionAndStore(key1 string, key2 string) bool { + ctx := context.Background() + err := RedisClient.SUnionStore(ctx, key1, key1, key2).Err() //将key1和key2的并集存入key1 + if err != nil { + fmt.Println("Error setting key: %v", err) + return false + } + return true +} + +// redis 清空set +func ClearRedisSet(key string) bool { + ctx := context.Background() + err := RedisClient.Del(ctx, key).Err() + if err != nil { + fmt.Println("Error setting key: %v", err) + return false + } + return true +} From 71783fe15f34671174e7b08260ab682362f1b55c Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Fri, 13 Dec 2024 20:26:39 +0800 Subject: [PATCH 5/6] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E7=A1=AE=E8=AE=A4=E9=83=A8?= =?UTF-8?q?=E5=88=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- service/userService.go | 21 +++++++++++---------- worker/redis.go | 11 +++++++++++ 2 files changed, 22 insertions(+), 10 deletions(-) diff --git a/service/userService.go b/service/userService.go index 0a7985e..db86c03 100644 --- a/service/userService.go +++ b/service/userService.go @@ -131,9 +131,13 @@ func UserSyncDataFromMaster() { // 同步数据到主服务器-增删改数据 func GetUserSyncData(device string) proto.UserSyncResp { key := device + "_sync_user_ids" - add_user_ids := worker.GetRedisSetMembers(key + "_add") - update_user_ids := worker.GetRedisSetMembers(key + "_update") - delete_user_ids := worker.GetRedisSetMembers(key + "_delete") + add_temp_key := device + "_sync_user_ids_add_confirm_temp" + update_temp_key := device + "_sync_user_ids_update_confirm_temp" + delete_temp_key := device + "_sync_user_ids_delete_confirm_temp" + //需要获取暂存集合的并集，清空暂存集合，存入待确认集合 + add_user_ids := worker.GetRedisSetUnion(key+"_add", add_temp_key) + update_user_ids := worker.GetRedisSetUnion(key+"_update", update_temp_key) + delete_user_ids := worker.GetRedisSetUnion(key+"_delete", delete_temp_key) add_users := []dao.User{} update_users := []dao.User{} delete_users := []proto.UserDelID{} @@ -153,16 +157,13 @@ func GetUserSyncData(device string) proto.UserSyncResp { id, _ := strconv.Atoi(v) delete_users = append(delete_users, proto.UserDelID{ID: uint(id)}) } - //将id存入暂存集合，清空原集合 - add_temp_key := device + "_sync_user_ids_add_confirm_temp" - update_temp_key := device + "_sync_user_ids_update_confirm_temp" - delete_temp_key := device + "_sync_user_ids_delete_confirm_temp" + //将id存入暂存集合，清空原集合，存入待确认集合主要保证在确认时，有新的数据加入不会在确认时漏掉 worker.SetRedisSetUnionAndStore(add_temp_key, key+"_add") - worker.SetRedisSetClear(key + "_add") + worker.ClearRedisSet(key + "_add") worker.SetRedisSetUnionAndStore(update_temp_key, key+"_update") - worker.SetRedisSetClear(key + "_update") + worker.ClearRedisSet(key + "_update") worker.SetRedisSetUnionAndStore(delete_temp_key, key+"_delete") - worker.SetRedisSetClear(key + "_delete") + worker.ClearRedisSet(key + "_delete") return proto.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users} } diff --git a/worker/redis.go b/worker/redis.go index 84bc598..ae83f49 100644 --- a/worker/redis.go +++ b/worker/redis.go @@ -449,3 +449,14 @@ func ClearRedisSet(key string) bool { } return true } + +// 获取两个集合的并集 +func GetRedisSetUnion(key1 string, key2 string) []string { + ctx := context.Background() + val, err := RedisClient.SUnion(ctx, key1, key2).Result() + if err != nil { + fmt.Println("Error getting key: %v", err) + return nil + } + return val +} From 62d336e1c0d39eca5111a9e28f7be8de835fbaf8 Mon Sep 17 00:00:00 2001 From: junleea <354425203@qq.com> Date: Sun, 15 Dec 2024 14:54:43 +0800 Subject: [PATCH 6/6] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E7=94=A8=E6=88=B7?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5=E9=85=8D=E7=BD=AE=E6=96=87?= =?UTF-8?q?=E4=BB=B6=E8=AF=BB=E5=8F=96=E5=8F=8A=E9=85=8D=E7=BD=AE=E6=96=B9?= =?UTF-8?q?=E5=BC=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main.go | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/main.go b/main.go index a2114e3..0254540 100644 --- a/main.go +++ b/main.go @@ -224,21 +224,30 @@ func ReadConfigToSetSystem() { cron_infos = append(cron_infos, logClean) } } - if proto.Config.SERVER_USER_TYPE == "slave" && proto.Config.USER_SYNC_TIME > 0 { - var is_exist bool - for _, v := range cron_infos { - if v.Type == 2 { - is_exist = true - break + + var is_exist bool + user_sync_id := -1 + for i, v := range cron_infos { + if v.Type == 2 { + is_exist = true + if proto.Config.USER_SYNC_TIME != v.Every { + v.Every = proto.Config.USER_SYNC_TIME + v.Curr = proto.Config.USER_SYNC_TIME } + user_sync_id = i + break } - if !is_exist { + } + if proto.Config.SERVER_USER_TYPE == "slave" { + if proto.Config.USER_SYNC_TIME > 0 && !is_exist { var userSync proto.CronInfo userSync.Type = 2 userSync.Info = "user" userSync.Curr = proto.Config.USER_SYNC_TIME userSync.Every = proto.Config.USER_SYNC_TIME cron_infos = append(cron_infos, userSync) + } else if user_sync_id != -1 { + cron_infos = append(cron_infos[:user_sync_id], cron_infos[user_sync_id+1:]...) //删除 } }