package task_pg import ( "fmt" "sync" "time" "github.com/hibiken/asynq" "go.uber.org/zap" ) type SchedulerManager struct { utc *asynq.Scheduler // UTC时间 zh *asynq.Scheduler // 中国时间(上海) sg *asynq.Scheduler // 新加坡时间 la *asynq.Scheduler // 洛杉矶时间 ny *asynq.Scheduler // 纽约时间 } var once sync.Once var schedulerManager SchedulerManager func NewSchedulerManager(redisOpt asynq.RedisClientOpt) *SchedulerManager { once.Do(func() { zh_loc, err := time.LoadLocation("Asia/Shanghai") if err != nil { zap.L().Error("Load timezone Asia/Shanghai failed.", zap.String("err", err.Error())) } sg_loc, err := time.LoadLocation("Asia/Singapore") if err != nil { zap.L().Error("Load timezone Asia/Singapore failed.", zap.String("err", err.Error())) } la_loc, err := time.LoadLocation("America/Los_Angeles") if err != nil { zap.L().Error("Load timezone America/Los_Angeles failed.", zap.String("err", err.Error())) } ny_loc, err := time.LoadLocation("America/New_York") if err != nil { zap.L().Error("Load timezone America/New_York failed.", zap.String("err", err.Error())) } schedulerManager = SchedulerManager{ utc: asynq.NewScheduler(redisOpt, &asynq.SchedulerOpts{Location: time.UTC}), zh: asynq.NewScheduler(redisOpt, &asynq.SchedulerOpts{Location: zh_loc}), sg: asynq.NewScheduler(redisOpt, &asynq.SchedulerOpts{Location: sg_loc}), la: asynq.NewScheduler(redisOpt, &asynq.SchedulerOpts{Location: la_loc}), ny: asynq.NewScheduler(redisOpt, &asynq.SchedulerOpts{Location: ny_loc}), } schedulerManager.utc.Start() schedulerManager.zh.Start() schedulerManager.sg.Start() schedulerManager.la.Start() schedulerManager.ny.Start() }) return &schedulerManager } // 注册UTC Scheduler func (sm *SchedulerManager) RegisterUtcScheduler(name string, cronspec string, task *asynq.Task, opts ...asynq.Option) error { if sm.utc == nil { return fmt.Errorf("RegisterUtcScheduler failed, SchedulerManager utc not initialized.(%s)", name) } entryID, err := sm.utc.Register(cronspec, task, opts...) if err != nil { zap.L().Error("UTC Scheduler register failed.", zap.String("err", err.Error()), zap.String("name", name)) return err } zap.L().Info("UTC Scheduler register successful.", zap.String("name", name), zap.String("entryID", entryID)) return nil } // 注册ZH(China) Scheduler func (sm *SchedulerManager) RegisterZhScheduler(name string, cronspec string, task *asynq.Task, opts ...asynq.Option) error { if sm.zh == nil { return fmt.Errorf("RegisterUtcScheduler failed, SchedulerManager zh not initialized.") } entryID, err := sm.zh.Register(cronspec, task, opts...) if err != nil { zap.L().Error("ZH Scheduler register failed.", zap.String("err", err.Error()), zap.String("name", name)) return err } zap.L().Info("ZH Scheduler register successful.", zap.String("name", name), zap.String("entryID", entryID)) return nil } // 注册SG(Singapore) Scheduler func (sm *SchedulerManager) RegisterSgScheduler(name string, cronspec string, task *asynq.Task, opts ...asynq.Option) error { if sm.sg == nil { return fmt.Errorf("RegisterUtcScheduler failed, SchedulerManager sg not initialized.") } entryID, err := sm.sg.Register(cronspec, task, opts...) if err != nil { zap.L().Error("SG Scheduler register failed.", zap.String("err", err.Error()), zap.String("name", name)) return err } zap.L().Info("SG Scheduler register successful.", zap.String("name", name), zap.String("entryID", entryID)) return nil } // 注册LA(US/Los Angeles) Scheduler func (sm *SchedulerManager) RegisterLaScheduler(name string, cronspec string, task *asynq.Task, opts ...asynq.Option) error { if sm.la == nil { return fmt.Errorf("RegisterUtcScheduler failed, SchedulerManager la not initialized.") } entryID, err := sm.la.Register(cronspec, task, opts...) if err != nil { zap.L().Error("LA Scheduler register failed.", zap.String("err", err.Error()), zap.String("name", name)) return err } zap.L().Info("LA Scheduler register successful.", zap.String("name", name), zap.String("entryID", entryID)) return nil } // 注册NY(US/New York) Scheduler func (sm *SchedulerManager) RegisterNyScheduler(name string, cronspec string, task *asynq.Task, opts ...asynq.Option) error { if sm.ny == nil { return fmt.Errorf("RegisterUtcScheduler failed, SchedulerManager ny not initialized.") } entryID, err := sm.ny.Register(cronspec, task, opts...) if err != nil { zap.L().Error("NY Scheduler register failed.", zap.String("err", err.Error()), zap.String("name", name)) return err } zap.L().Info("NY Scheduler register successful.", zap.String("name", name), zap.String("entryID", entryID)) return nil }