优化结构
This commit is contained in:
@@ -75,7 +75,8 @@ func TestReloadPreservesExistingScheduleOnPreparationFailure(t *testing.T) {
|
||||
repo := &workerTaskRepo{}
|
||||
scheduler := newTestTaskScheduler(repo)
|
||||
old := &taskbiz.TimedTask{ID: 1, Name: "old", Spec: "0 0 * * *", Enabled: true}
|
||||
if err := scheduler.Schedule(old); err != nil {
|
||||
repo.items = []*taskbiz.TimedTask{old}
|
||||
if err := scheduler.ScheduleID(context.Background(), old.ID); err != nil {
|
||||
t.Fatalf("schedule old task: %v", err)
|
||||
}
|
||||
|
||||
@@ -116,16 +117,17 @@ func assertOnlyScheduledTask(t *testing.T, scheduler *TaskScheduler, id uint) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestConcurrentScheduleKeepsSingleEntry(t *testing.T) {
|
||||
scheduler := newTestTaskScheduler(&workerTaskRepo{})
|
||||
func TestConcurrentScheduleIDKeepsSingleEntry(t *testing.T) {
|
||||
repo := &workerTaskRepo{items: []*taskbiz.TimedTask{{ID: 7, Name: "same", Spec: "0 0 * * *", Enabled: true}}}
|
||||
scheduler := newTestTaskScheduler(repo)
|
||||
const workers = 32
|
||||
var wg sync.WaitGroup
|
||||
for i := 0; i < workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
if err := scheduler.Schedule(&taskbiz.TimedTask{ID: 7, Name: "same", Spec: "0 0 * * *", Enabled: true}); err != nil {
|
||||
t.Errorf("Schedule() error = %v", err)
|
||||
if err := scheduler.ScheduleID(context.Background(), 7); err != nil {
|
||||
t.Errorf("ScheduleID() error = %v", err)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
@@ -325,34 +325,6 @@ func (s *TaskScheduler) run(ctx context.Context, task *taskbiz.TimedTask, trigge
|
||||
}
|
||||
}
|
||||
|
||||
func (s *TaskScheduler) Schedule(task *taskbiz.TimedTask) error {
|
||||
if task == nil {
|
||||
return errors.New("定时任务不能为空")
|
||||
}
|
||||
var schedule cron.Schedule
|
||||
var err error
|
||||
if task.Enabled {
|
||||
schedule, err = parseTaskSchedule(task)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
s.runMu.Lock()
|
||||
stopping := s.stopping
|
||||
s.runMu.Unlock()
|
||||
if stopping {
|
||||
return errors.New("任务调度器正在停止")
|
||||
}
|
||||
s.removeLocked(task.ID)
|
||||
if !task.Enabled {
|
||||
return nil
|
||||
}
|
||||
s.scheduleLocked(cloneTimedTask(task), schedule)
|
||||
return nil
|
||||
}
|
||||
|
||||
func parseTaskSchedule(task *taskbiz.TimedTask) (cron.Schedule, error) {
|
||||
if task.WithSeconds {
|
||||
return cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor).Parse(task.Spec)
|
||||
|
||||
Reference in New Issue
Block a user