Merge branch 'refs/heads/feat-user-sync' into release

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) {
+func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) error {
-	DB.Model(&User{}).Where("id = ?", id).Updates(User{Name: req.Username, Age: req.Age, Avatar: req.Avatar, Gender: req.Gender})
+	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"` //删除用户
 }

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
+	}
+}

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))
 		}
-
 	}
 }
 

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
+	DB                   int    `json:"db"` // 0: mysql, 1: pg
-	MYSQL_DSN       string `json:"mysql_dsn"`
+	MYSQL_DSN            string `json:"mysql_dsn"`
-	PG_DSN          string `json:"pg_dsn"`
+	PG_DSN               string `json:"pg_dsn"`
-	REDIS_ADDR      string `json:"redis_addr"`
+	REDIS_ADDR           string `json:"redis_addr"`
-	TOKEN_USE_REDIS bool   `json:"token_use_redis"`
+	TOKEN_USE_REDIS      bool   `json:"token_use_redis"`
-	REDIS_User_PW   bool   `json:"redis_user_pw"` // 是否使用密码
+	REDIS_User_PW        bool   `json:"redis_user_pw"` // 是否使用密码
-	REDIS_PASSWORD  string `json:"redis_password"`
+	REDIS_PASSWORD       string `json:"redis_password"`
-	REDIS_DB        int    `json:"redis_db"`
+	REDIS_DB             int    `json:"redis_db"`
-	TOKEN_SECRET    string `json:"token_secret"`
+	TOKEN_SECRET         string `json:"token_secret"`
-	CID_BASE_DIR    string `json:"cid_base_dir"`
+	CID_BASE_DIR         string `json:"cid_base_dir"`
-	FILE_BASE_DIR   string `json:"file_base_dir"`
+	FILE_BASE_DIR        string `json:"file_base_dir"`
-	MONITOR         bool   `json:"monitor"`       // 状态监控及邮件通知
+	MONITOR              bool   `json:"monitor"`              // 状态监控及邮件通知
-	SERVER_PORT     string `json:"server_port"`   // 服务端口
+	SERVER_PORT          string `json:"server_port"`          // 服务端口
-	LOG_SAVE_DAYS   int    `json:"log_save_days"` // 日志保存天数,-1表示不保存，0表示永久保存
+	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"`          // 服务器名称,用于区分不同服务器
 }
 
 // 读取配置文件

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"`
+}

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) //用户修改自己的信息，不能修改权限信息
+		err := dao.UpdateUserByID3(user_id, req) //用户修改自己的信息，不能修改权限信息
-		return user_id, nil
+		//添加修改用户信息到同步列表
+		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
+}

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
+}

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
+}