diff --git a/dao/user.go b/dao/user.go index 5f10bbe..87f64ea 100644 --- a/dao/user.go +++ b/dao/user.go @@ -112,7 +112,11 @@ func UpdateUserByID3(id int, req proto.UpdateUserInfoReq) 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) + 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 + } + res = DB.Debug().Exec("update users set deleted_at=null where id=?", req.ID) if res.Error != nil { return 0 } @@ -121,10 +125,15 @@ func AddUserSync(req proto.UserAddOrUpdate) uint { // 用户数据同步-更新 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 } + res = DB.Debug().Exec("update users set deleted_at=null where id=?", req.ID) + if res.Error != nil { + return 0 + } return req.ID } @@ -150,3 +159,9 @@ type UserSyncResp struct { Add []User `json:"add" form:"add"` //添加用户 Delete []proto.UserDelID `json:"delete" form:"delete"` //删除用户 } + +// 清空用户表 +func ClearAllUsers() error { + res := DB.Exec("TRUNCATE TABLE users") + return res.Error +} diff --git a/main.go b/main.go index 0254540..10c9fb7 100644 --- a/main.go +++ b/main.go @@ -9,6 +9,7 @@ import ( "io" "log" "os" + "strconv" "strings" "videoplayer/dao" "videoplayer/handler" @@ -142,7 +143,10 @@ func JWTAuthMiddleware() gin.HandlerFunc { }) return } - c.Set("id", s_id) + id, _ := strconv.Atoi(s_id) + id_float64 := float64(id) + //查看s_id类型 + c.Set("id", id_float64) c.Next() return } @@ -225,8 +229,8 @@ func ReadConfigToSetSystem() { } } - var is_exist bool - user_sync_id := -1 + is_exist := false + user_sync_id := -1 //用户同步任务索引 for i, v := range cron_infos { if v.Type == 2 { is_exist = true @@ -235,6 +239,7 @@ func ReadConfigToSetSystem() { v.Curr = proto.Config.USER_SYNC_TIME } user_sync_id = i + cron_infos[i] = v break } } @@ -289,7 +294,7 @@ func RunGeneralCron() { if err != nil { fmt.Println("RunGeneralCron Error decoding config,key value is :", res) } - for _, v := range cron_infos { + for i, v := range cron_infos { //1:日志清理,其他待定 if v.Type == 1 { //日志清理 @@ -300,6 +305,8 @@ func RunGeneralCron() { } else { v.Curr -= 10 } + cron_infos[i] = v + continue } //2 从服务器同步数据 if v.Type == 2 { @@ -312,6 +319,8 @@ func RunGeneralCron() { } else { v.Curr -= 10 } + cron_infos[i] = v + continue } } //存入redis diff --git a/service/userService.go b/service/userService.go index 41cfb00..ad44cb6 100644 --- a/service/userService.go +++ b/service/userService.go @@ -112,14 +112,27 @@ func UserSyncDataFromMaster() { req.Token = tokens[0] req.Device = proto.Config.SERVER_NAME all := worker.GetRedis("user_sync_all") + is_all := false //是否全量同步 if all == "" || all == "1" { + is_all = true + //清空数据表 + err := dao.ClearAllUsers() + if err != nil { + fmt.Println("All ClearAllUsers error:", err) + return + } + worker.SetRedis("user_sync_all", "1") req.Types = 1 } else { worker.SetRedis("user_sync_all", "2") req.Types = 2 } - user_sync_data := worker.SyncDataFromMasterReq2(url, req) + user_sync_data, err := worker.SyncDataFromMasterReq2(url, req) + if err != nil { + fmt.Println("UserSyncDataFromMaster error:", err) + return + } add_users := user_sync_data.Add update_users := user_sync_data.Update delete_users := user_sync_data.Delete @@ -158,17 +171,21 @@ func UserSyncDataFromMaster() { } //确认同步数据 - 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) + if is_all == false { + 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) + } else { + worker.SetRedis("user_sync_all", "2") + } } // 同步数据到主服务器-增删改数据 diff --git a/worker/req.go b/worker/req.go index fcbefd4..a8a30a2 100644 --- a/worker/req.go +++ b/worker/req.go @@ -129,17 +129,29 @@ func SyncDataFromMasterReq(url string, token string) proto.UserSync { return userSync } +type Response struct { + Code int `json:"code"` + Message string `json:"message"` + Data proto.UserSync `json:"data"` +} + // 获取数据,全量及增量 -func SyncDataFromMasterReq2(url string, data proto.SyncUserReq) proto.UserSync { +func SyncDataFromMasterReq2(url string, data proto.SyncUserReq) (proto.UserSync, error) { + defer func() { + if r := recover(); r != nil { + fmt.Println("SyncDataFromMasterReq2 error:", r) + } + }() + var res proto.UserSync //从接口获取数据 json_data, err := json.Marshal(data) if err != nil { - return res + return res, err } req, err := http.NewRequest("POST", url, bytes.NewBuffer(json_data)) if err != nil { - return res + return res, err } req.Header.Set("Content-Type", "application/json") //传输数据 @@ -149,22 +161,20 @@ func SyncDataFromMasterReq2(url string, data proto.SyncUserReq) proto.UserSync { //获取数据 resp, err := client.Do(req) if err != nil { - return res + return res, err } defer resp.Body.Close() //解析数据 - var m map[string]interface{} - err = json.NewDecoder(resp.Body).Decode(&m) + responseBod, err := io.ReadAll(resp.Body) if err != nil { - return res + return res, err } - if m["code"].(float64) != 0 { - return res - } - err = json.Unmarshal([]byte(m["data"].(string)), &res) + var response Response + err = json.Unmarshal(responseBod, &response) if err != nil { - fmt.Println("SyncDataFromMasterReq2 error decode data:", err) - return res + return res, err } - return res + res = response.Data + fmt.Println("SyncDataFromMasterReq2 result add data:", len(res.Add), "update data:", len(res.Update), "delete data:", len(res.Delete)) + return res, nil }