FranklinWH-Toolbox/service/task/plugin/scheduler.go

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
}