package fwhQueue import ( "context" "fmt" "fwh-toolbox-server/domain" "fwh-toolbox-server/dto" "fwh-toolbox-server/logger" "fwh-toolbox-server/repositories" "go.uber.org/zap" ) var ( TaskTypeSiteInfoUpdate = "task_SiteInfoUpdate" ) // type TaskSiteInfoUpdate = types.TaskSiteInfoUpdate func handleTaskSiteInfoUpdate(ctx context.Context, payload domain.TaskSiteInfoUpdate) error { // 更新运行状态 fmt.Println("更新任务运行状态") if repoErr := repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressRunning, 0); repoErr != nil { zap.L().Error("更新站点信息Update结果状态失败(Running)", zap.String("err", repoErr.Error())) return repoErr } fmt.Printf("开始执行任务 %v\n", payload) total, err := executeTaskSiteInfoUpdate(ctx, payload) // 记录任务执行结果 if err != nil { fmt.Printf("任务执行失败 %s\n", err.Error()) if repoErr := repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressError, 0); repoErr != nil { zap.L().Error("更新站点信息Update结果状态失败(Error)", zap.String("err", repoErr.Error())) return repoErr } } else { fmt.Println("任务执行成功") if repoErr := repositories.UpdateSiteInfoUpdateResult(payload.Id, domain.ProgressComplete, total); repoErr != nil { zap.L().Error("更新站点信息Update结果状态失败(Complete)", zap.String("err", repoErr.Error())) return repoErr } } return err } var executeTaskSiteInfoUpdate = func(ctx context.Context, payload domain.TaskSiteInfoUpdate) (uint64, error) { var pageSize uint = 50 var pageNum uint = 1 var total uint64 = 0 var realTotal uint64 = 0 fmt.Println("查询站点信息") for { resp, err := repositories.QuerySiteInfoList(dto.SiteInfoPara{ PageSize: pageSize, PageNum: pageNum, }) if err != nil { zap.L().Error("查询站点信息失败", zap.String("err", err.Error()), zap.Uint("pageSize", pageSize), zap.Uint("pageNum", pageNum), zap.Uint64("total", total)) return 0, err } for _, info := range *resp.Result { if err := repositories.InsertSiteInfoData(*domain.FwhSiteInfoToSiteInfoDomain(&info)); err != nil { zap.L().Error("保存站点信息失败", zap.String("err", err.Error()), zap.Uint("pageSize", pageSize), zap.Uint("pageNum", pageNum), zap.Uint64("total", total)) return 0, err } realTotal++ } pageNum++ total = resp.Total if (int64(pageNum)-1)*int64(pageSize) >= int64(total) { break } } return realTotal, nil } // 插入站点分析队列 func EnqueueSiteInfoUpdateTask(payload domain.TaskSiteInfoUpdate) error { p := TaskPayload[domain.TaskSiteInfoUpdate]{ TaskPayloadBase: TaskPayloadBase{ Type: TaskTypeSiteInfoUpdate, }, Payload: payload, } if err := EnqueueUniqueTask[domain.TaskSiteInfoUpdate](p); err != nil { logger.Std.Printf("EnqueueSiteInfoUpdateTask error : %s\n", err.Error()) return err } return nil }