package action import ( "context" "joybuywork/internal/app" "joybuywork/internal/consts" "joybuywork/internal/plugins/printer" "joybuywork/internal/repository/dao" "joybuywork/internal/repository/model" "joybuywork/internal/utils" "joybuywork/internal/utils/merr" "fmt" "time" "github.com/rs/xid" rl "gits.branchcn.com/backend/microcode/plugins/derivative/lock" ) const ( loggerModPrint = "print" loggerModDb = "db" ) type OnceKey struct { exp time.Duration //锁过期时间 key string //锁key release bool //是否直接完毕释放,(有一些可能需要一直锁着) } type SyncProgress struct { total int64 //任务总数 sp chan int64 //当前已处理的任务数 tf func() int64 //计算任务总数的方法 } type IAction interface { Init(ctx context.Context) //基于ctx的初始化处理 Exec() error //执行方法 OnceKey() *OnceKey //锁,防止多服务器并发操作 Desc() string //描述,用来做关键字搜索 Alert() bool //推送消息 LoggerMod() string //日志模式 DbExec() (total, successNum, failNum int64, err error) //Db执行方法 } type Action struct { ctx context.Context be *merr.BusinessError } func (a *Action) Exec() error { return nil } func (a *Action) DbExec() (total, successNum, failNum int64, err error) { return } func (a *Action) BE() *merr.BusinessError { return a.be } func (a *Action) Alert() bool { return false } func (a *Action) LoggerMod() string { return loggerModPrint } func (a *Action) Init(ctx context.Context) { a.ctx = ctx a.be = merr.NewBE(ctx) } type Job struct { R IAction } func (s *Job) Run() { taskId := xid.New().String() ctx, cancel := context.WithCancel(context.WithValue(context.Background(), consts.TraceIdKey, taskId)) defer cancel() s.R.Init(ctx) switch s.R.LoggerMod() { case loggerModPrint: s.print(ctx, taskId) case loggerModDb: s.db(ctx) } } func (s *Job) db(ctx context.Context) { //防止多服务器同时执行,用redis上锁 if s.R.OnceKey() != nil { lock := app.NewRedisLock(ctx, s.R.OnceKey().key, rl.SetExpire(s.R.OnceKey().exp)) if !lock.SoftLock() { return } if s.R.OnceKey().release { defer lock.Release() } } //执行任务 t := time.Now() total, successNum, failNum, err := s.R.DbExec() if total == 0 { return } dao.CronLog.WithContext(ctx).Create(&model.CronLog{ ExecAt: t.Unix(), Desc: s.R.Desc(), Machine: app.GetLocalIP(), Total: total, SuccessNum: successNum, FailNum: failNum, ExecStatus: utils.Ternary[int8](err == nil, 1, 2), ExecFail: utils.Ternary[string](err == nil, "", fmt.Sprintf("%v", err)), EndAt: t.Unix(), }) } func (s *Job) print(ctx context.Context, taskId string) { mark := s.R.Desc() //捕捉运行panic,不用默认的捕捉,因为没有链路追踪 defer func() { if err := recover(); err != nil { app.Logger.Error(ctx, fmt.Sprintf("[%s] 执行错误: %v", mark, err)) } if s.R.Alert() { app.FeishuNormalSend("task_id: %s, [%s]正常执行", taskId, mark) } }() //防止多服务器同时执行,用redis上锁 if s.R.OnceKey() != nil { lock := app.NewRedisLock(ctx, s.R.OnceKey().key, rl.SetExpire(s.R.OnceKey().exp)) if !lock.SoftLock() { return } if s.R.OnceKey().release { defer lock.Release() } } //打印日志,知道这个定时任务执行了 printer.Print(ctx, fmt.Sprintf("执行定时任务:%s", mark), nil) //执行任务 t := time.Now() err := s.R.Exec() if err != nil { app.Logger.Error(ctx, fmt.Sprintf("[%s] 执行错误: %v", mark, err)) return } //打印日志,知道这个定时任务执行了 printer.Print(ctx, fmt.Sprintf("[%s]执行完成,执行时长: %12v", mark, time.Since(t)), err) } func NewJob(r IAction) *Job { return &Job{R: r} }