diff --git a/dao/user.go b/dao/user.go index 1f8d702..5f10bbe 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 } @@ -81,7 +84,7 @@ func UpdateUserByID(id int, name, password, email string) { } // 管理员修改用户信息 -func UpdateUserByID2(id int, req proto.UpdateUserInfoReq) { +func UpdateUserByID2(id int, req proto.UpdateUserInfoReq) error { updateData := make(map[string]interface{}) updateData["Name"] = req.Username updateData["Age"] = req.Age @@ -94,10 +97,56 @@ func UpdateUserByID2(id int, req proto.UpdateUserInfoReq) { updateData["CIDFunc"] = req.CIDFunc updateData["Avatar"] = req.Avatar updateData["Gender"] = req.Gender - DB.Model(&User{}).Where("id =?", id).Updates(updateData) + res := DB.Model(&User{}).Where("id =?", id).Updates(updateData) + if res.Error != nil { + return res.Error + } + return nil } // 用户修改自己的信息 -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}) + return res.Error +} + +// 用户数据同步-添加 +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 +} + +// 获取所有用户 +func GetAllUser() []User { + var users []User + DB.Find(&users) + return users +} + +// 用户数据同步 +type UserSyncResp struct { + Update []User `json:"update" form:"update"` //更新用户 + Add []User `json:"add" form:"add"` //添加用户 + Delete []proto.UserDelID `json:"delete" form:"delete"` //删除用户 } diff --git a/handler/user.go b/handler/user.go index 11023d0..0c40dcb 100644 --- a/handler/user.go +++ b/handler/user.go @@ -26,6 +26,8 @@ func SetUpUserGroup(router *gin.Engine) { userGroup.POST("/search", SearchHandler) userGroup.POST("/info", GetUserInfo) userGroup.POST("/update", UpdateUserInfo) + userGroup.POST("/sync", GetSyncUserInfo) + userGroup.POST("/delete", DeleteUser) } type RLReq struct { @@ -77,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") @@ -350,3 +369,49 @@ func registerHandler(c *gin.Context) { c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": data}) return } + +func GetSyncUserInfo(c *gin.Context) { + var req_data proto.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 { //1为全量同步 + add_users := dao.GetAllUser() + resp := dao.UserSyncResp{} + resp.Add = add_users + c.JSON(200, gin.H{"code": proto.SuccessCode, "message": "success", "data": resp}) + } 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 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:" + res.Error(), "data": "failed"}) + } + } else { + c.JSON(200, gin.H{"code": proto.ParameterError, "message": "type is error", "data": dao.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/main.go b/main.go index 186785d..0254540 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,34 @@ func ReadConfigToSetSystem() { logClean.Every = 86400 cron_infos = append(cron_infos, logClean) } - } + + 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 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:]...) //删除 + } + } + } else { if proto.Config.LOG_SAVE_DAYS > 0 { var logClean proto.CronInfo @@ -233,6 +260,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 +301,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 +321,6 @@ func RunGeneralCron() { } else { worker.SetRedis(key, string(json_data)) } - } } diff --git a/proto/conf.go b/proto/conf.go index d014382..e6d5670 100644 --- a/proto/conf.go +++ b/proto/conf.go @@ -9,7 +9,7 @@ import ( var Config ConfigStruct var SigningKey = []byte{} -var Url_map = map[string]bool{"/login": true, "/register": true, "/uuid": true, "/gqr": true, "/cid/callback": true, "/tool/monitor": true} // 不需要token验证的url +var Url_map = map[string]bool{"/login": true, "/register": true, "/uuid": true, "/gqr": true, "/cid/callback": true, "/tool/monitor": true, "/user/sync": true} // 不需要token验证的url var Per_menu_map = map[string]int{"/video/": 1, "/device/": 2, "/cid/": 3} const ( @@ -63,20 +63,24 @@ 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"` // 用户数据同步时间，单位秒 + SERVER_NAME string `json:"server_name"` // 服务器名称,用于区分不同服务器 } // 读取配置文件 diff --git a/proto/user_req.go b/proto/user_req.go index a42f24b..9d06f03 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,60 @@ 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 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 +} + +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"` +} + +// 数据同步请求 +type SyncUserReq struct { + Token string `json:"token" form:"token"` + Types int `json:"type" form:"type"` // 1为全量同步 2为增量同步 + Device string `json:"device" form:"device"` + Confirm UserSyncConfirm `json:"confirm" form:"confirm"` +} diff --git a/service/userService.go b/service/userService.go index b395b94..41cfb00 100644 --- a/service/userService.go +++ b/service/userService.go @@ -1,13 +1,26 @@ package service import ( + "errors" + "fmt" "regexp" + "strconv" + "time" "videoplayer/dao" "videoplayer/proto" + "videoplayer/worker" ) 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 { @@ -46,12 +59,244 @@ 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 { + fmt.Println("set sync user data set error:", err2) + return user_id, nil + } + } + return user_id, err } else if cur_user.Role == "admin" { - dao.UpdateUserByID2(req.ID, req) + err := dao.UpdateUserByID2(req.ID, req) + if err == nil { + //添加修改用户信息到同步列表 + err2 := setSyncUserDataSet("update", req.ID) + if err2 != nil { + fmt.Println("set sync user data set error:", err2) + return req.ID, nil + } + } return req.ID, nil } else { return 0, nil } } + +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 := "https://" + proto.Config.MASTER_SERVER_DOMAIN + "/user/sync" + tokens := worker.GetRedisSetMembers("super_permission_tokens") + var req proto.SyncUserReq + req.Token = tokens[0] + req.Device = proto.Config.SERVER_NAME + all := worker.GetRedis("user_sync_all") + if all == "" || all == "1" { + req.Types = 1 + } else { + worker.SetRedis("user_sync_all", "2") + req.Types = 2 + } + + user_sync_data := worker.SyncDataFromMasterReq2(url, req) + add_users := user_sync_data.Add + update_users := user_sync_data.Update + delete_users := user_sync_data.Delete + //未成功操作的id + var fail_ids []uint + + //添加用户 + var add_confirm []proto.UserConfirmID + for _, v := range add_users { + res := dao.AddUserSync(v) + if res == 0 { + fail_ids = append(fail_ids, v.ID) + } else { + add_confirm = append(add_confirm, proto.UserConfirmID{ID: v.ID}) + } + } + //更新用户 + var update_confirm []proto.UserConfirmID + for _, v := range update_users { + res := dao.UpdateUserSync(v) + if res == 0 { + fail_ids = append(fail_ids, v.ID) + } else { + update_confirm = append(update_confirm, proto.UserConfirmID{ID: v.ID}) + } + } + //删除用户 + var delete_confirm []proto.UserConfirmID + for _, v := range delete_users { + res := dao.DeleteUserSync(v) + if res == 0 { + fail_ids = append(fail_ids, v.ID) + } else { + delete_confirm = append(delete_confirm, proto.UserConfirmID{ID: v.ID}) + } + } + + //确认同步数据 + var data proto.UserSyncConfirm + data.Add = add_confirm + data.Update = update_confirm + data.Delete = delete_confirm + //确认同步数据请求 + var confirm_req proto.SyncUserReq + confirm_req.Token = tokens[0] + confirm_req.Device = proto.Config.SERVER_NAME + confirm_req.Types = 3 + confirm_req.Confirm = data + worker.SyncDataFromMasterReq2(url, confirm_req) +} + +// 同步数据到主服务器-增删改数据 +func GetUserSyncData(device string) dao.UserSyncResp { + key := device + "_sync_user_ids" + 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{} + 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)}) + } + //将id存入暂存集合，清空原集合，存入待确认集合主要保证在确认时，有新的数据加入不会在确认时漏掉 + worker.SetRedisSetUnionAndStore(add_temp_key, key+"_add") + worker.ClearRedisSet(key + "_add") + worker.SetRedisSetUnionAndStore(update_temp_key, key+"_update") + worker.ClearRedisSet(key + "_update") + worker.SetRedisSetUnionAndStore(delete_temp_key, key+"_delete") + worker.ClearRedisSet(key + "_delete") + return dao.UserSyncResp{Add: add_users, Update: update_users, Delete: delete_users} +} + +func setSyncUserDataSet(t string, id int) error { + devices := worker.GetRedisSetMembers("sync_devices_ids") //主服务器查看从服务器的设备列表 + fmt.Println("set sync user data set devices:", devices, "t:", t, "id:", id) + 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 +} + +// 确认同步数据 +func ConfirmSyncUserData(device string, data proto.UserSyncConfirm) error { + var err error + if len(data.Add) > 0 { + var ids_add []string + + 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 add confirm error") + return err + } + ids_add_confirm_temp := device + "_sync_user_ids_add_confirm_temp" + //取差集 + add_diff := worker.SetRedisSetDiffAndStore(ids_add_confirm_temp, add_key) + if add_diff == false { + err = errors.New("add diff error") + return err + } + } + + if len(data.Update) > 0 { + + 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 update confirm error") + return err + } + ids_update_confirm_temp := device + "_sync_user_ids_update_confirm_temp" + + update_diff := worker.SetRedisSetDiffAndStore(ids_update_confirm_temp, update_key) + if update_diff == false { + err = errors.New("update diff error") + return err + } + + } + + if len(data.Delete) > 0 { + 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 del confirm error") + return err + } + + //待确认集合暂存 + ids_delete_confirm_temp := device + "_sync_user_ids_delete_confirm_temp" + delete_diff := worker.SetRedisSetDiffAndStore(ids_delete_confirm_temp, del_key) + if delete_diff == false { + err = errors.New("delete diff error") + return err + } + } + return err +} diff --git a/worker/redis.go b/worker/redis.go index a9957bb..d551f3e 100644 --- a/worker/redis.go +++ b/worker/redis.go @@ -292,17 +292,34 @@ 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("SetRedisSetAddBatchWithExpire Error setting key: %v", err) + return false + } + err = RedisClient.Expire(ctx, key, expire).Err() + if err != nil { + fmt.Println("SetRedisSetAddBatchWithExpire Error setting key: %v", err) + return false + } + return true + +} + // 设置set,添加元素 func SetRedisSetAddWithExpire(key string, value string, expire time.Duration) bool { ctx := context.Background() err := RedisClient.SAdd(ctx, key, value).Err() if err != nil { - fmt.Println("Error setting key: %v", err) + fmt.Println("SetRedisSetAddWithExpire Error setting key: %v", err) return false } err = RedisClient.Expire(ctx, key, expire).Err() if err != nil { - fmt.Println("Error setting key: %v", err) + fmt.Println("SetRedisSetAddWithExpire Error setting key: %v", err) return false } return true @@ -399,3 +416,47 @@ 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("SetRedisSetDiffAndStore 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("SetRedisSetUnionAndStore 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 +} + +// 获取两个集合的并集 +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 +} diff --git a/worker/req.go b/worker/req.go index e271110..fcbefd4 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,89 @@ 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) + //json负载 + req.Header.Set("Content-Type", "application/json") + //传输数据 + m := make(map[string]interface{}) + m["token"] = token + m["device"] = "" + + if client == nil { + client = &http.Client{} + } + 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 +} + +// 获取数据,全量及增量 +func SyncDataFromMasterReq2(url string, data proto.SyncUserReq) proto.UserSync { + var res proto.UserSync + //从接口获取数据 + json_data, err := json.Marshal(data) + if err != nil { + return res + } + req, err := http.NewRequest("POST", url, bytes.NewBuffer(json_data)) + if err != nil { + return res + } + req.Header.Set("Content-Type", "application/json") + //传输数据 + if client == nil { + client = &http.Client{} + } + //获取数据 + resp, err := client.Do(req) + if err != nil { + return res + } + defer resp.Body.Close() + //解析数据 + var m map[string]interface{} + err = json.NewDecoder(resp.Body).Decode(&m) + if err != nil { + return res + } + if m["code"].(float64) != 0 { + return res + } + err = json.Unmarshal([]byte(m["data"].(string)), &res) + if err != nil { + fmt.Println("SyncDataFromMasterReq2 error decode data:", err) + return res + } + return res +}