// 用户定时任务调度 — Linux 标准 5 段 crontab(分 时 日 月 周)// Logan.Li package cronscheduler import ( "log" "strings" "sync" "time" "ezops/model" "github.com/robfig/cron/v3" "gorm.io/gorm" ) // Runner 执行定时任务回调(由 controller 注入)// Logan.Li type Runner func(jobID uint) type trackedEntry struct { entryID cron.EntryID expr string } // Scheduler 后台 Cron 调度器 // Logan.Li type Scheduler struct { db *gorm.DB runner Runner cron *cron.Cron mu sync.Mutex entries map[uint]trackedEntry running sync.Map } // Start 启动调度器:加载已启用任务,并周期性同步 DB 变更 // Logan.Li func Start(db *gorm.DB, runner Runner) *Scheduler { s := &Scheduler{ db: db, runner: runner, cron: cron.New(cron.WithLocation(time.Local)), entries: make(map[uint]trackedEntry), } s.cron.Start() s.reload() go func() { ticker := time.NewTicker(30 * time.Second) defer ticker.Stop() for range ticker.C { s.reload() } }() log.Println("[CronScheduler] 已启动(Linux 5 段表达式:分 时 日 月 周)") return s } func (s *Scheduler) Stop() { s.cron.Stop() } func (s *Scheduler) reload() { var jobs []model.CronJob if err := s.db.Where("enabled = ?", true).Find(&jobs).Error; err != nil { log.Printf("[CronScheduler] 加载任务失败: %v", err) return } s.mu.Lock() defer s.mu.Unlock() active := make(map[uint]struct{}, len(jobs)) for _, job := range jobs { expr := normalizeCronExpr(job.CronExpr) if expr == "" { if prev, ok := s.entries[job.ID]; ok { s.cron.Remove(prev.entryID) delete(s.entries, job.ID) } continue } active[job.ID] = struct{}{} if prev, ok := s.entries[job.ID]; ok && prev.expr == expr { continue } if prev, ok := s.entries[job.ID]; ok { s.cron.Remove(prev.entryID) } jobID := job.ID entryID, err := s.cron.AddFunc(expr, func() { s.dispatch(jobID) }) if err != nil { log.Printf("[CronScheduler] 注册任务 %s(%d) 失败 expr=%q: %v", job.Name, job.ID, expr, err) delete(s.entries, job.ID) continue } s.entries[job.ID] = trackedEntry{entryID: entryID, expr: expr} log.Printf("[CronScheduler] 已注册: %s id=%d expr=%s", job.Name, job.ID, expr) } for id, prev := range s.entries { if _, ok := active[id]; !ok { s.cron.Remove(prev.entryID) delete(s.entries, id) log.Printf("[CronScheduler] 已移除: id=%d", id) } } } func (s *Scheduler) dispatch(jobID uint) { if _, loaded := s.running.LoadOrStore(jobID, true); loaded { log.Printf("[CronScheduler] 任务 id=%d 仍在执行,跳过本次调度", jobID) return } defer s.running.Delete(jobID) if s.runner == nil { return } s.runner(jobID) } // normalizeCronExpr 规整为 Linux crontab 五段式 // Logan.Li func normalizeCronExpr(expr string) string { expr = strings.TrimSpace(expr) if expr == "" { return "" } fields := strings.Fields(expr) if len(fields) == 5 { return strings.Join(fields, " ") } // 误填 6 段(含秒)时去掉首段,兼容常见 Quartz 写法 // Logan.Li if len(fields) == 6 { return strings.Join(fields[1:], " ") } return "" }