update

config/config.go

@@ -72,6 +72,14 @@ func GetOssConfig() *oss.AliOssConfig {
 	return ossCfg
 }
 
+func GetCron() *Cron {
+	cron := Cron{}
+	Read(func(c *Config) {
+		copier.Copy(&cron, &c.Cron)
+	})
+	return &cron
+}
+
 func Read(f func(c *Config)) {
 	if Cfg == nil {
 		panic("config is nil, please call config.Init() first")

xcron/cron.go

@@ -110,14 +110,22 @@ func (x *cronJobController) CallOnceByCode(c *gin.Context) {
 	response.Success(c)
 }
 
-func Init(schema string, handlers map[string]Fn) error {
-	dao.Init(schema)
+func AddHandler(k string, h Fn) {
+	Handlers.Add(k, h)
+}
 
-	ctx := context.TODO()
-	TaskRegisterMaps = handlers
-	if !config.CronOpen() {
+func AddHandlerMap(handlers map[string]Fn) {
+	for k, h := range handlers {
+		AddHandler(k, h)
+	}
+}
+
+func initCron() error {
+	cronCfg := config.GetCron()
+	if !cronCfg.Enable {
 		return nil
 	}
+	ctx := context.TODO()
 	cronJobs, err := service.CronJobService.GetSliceByEnable(ctx, enums.CronJob_Enable_OPEN, &biz.CronJobOpts{})
 	if err != nil {
 		return err
@@ -130,7 +138,7 @@ func Init(schema string, handlers map[string]Fn) error {
 	var tasks []*gocron.Task
 	for _, job := range cronJobs {
 		job.ParseJobInfo()
-		_, ok := handlers[job.JobCode]
+		_, ok := Handlers.Get(job.JobCode)
 		if ok {
 			tsk := gocron.Task{
 				TimeSpec: job.TimeSpec,
@@ -162,13 +170,28 @@ func Init(schema string, handlers map[string]Fn) error {
 
 type Fn func(ctx context.Context, job *biz.CronJob) error
 
-var TaskRegisterMaps = map[string]Fn{}
+type handlers struct {
+	M map[string]Fn
+}
+
+var Handlers = &handlers{
+	M: make(map[string]Fn),
+}
+
+func (x *handlers) Get(key string) (Fn, bool) {
+	v, ok := x.M[key]
+	return v, ok
+}
+
+func (x *handlers) Add(key string, h Fn) {
+	x.M[key] = h
+}
 
 func ExecJob(ctx context.Context, j *biz.CronJob) error {
 	if j == nil {
 		return errors.New("job is nil")
 	}
-	_, ok := TaskRegisterMaps[j.JobCode]
+	_, ok := Handlers.Get(j.JobCode)
 	if !ok {
 		return errors.New("handler not found")
 	}
@@ -190,7 +213,7 @@ func execTask(ctx context.Context, j *biz.CronJob) error {
 	if j == nil {
 		return errors.New("job not found")
 	}
-	fn, ok := TaskRegisterMaps[j.JobCode]
+	fn, ok := Handlers.Get(j.JobCode)
 	if !ok {
 		return errors.New("handler not found")
 	}

xcron/dao/var.go

-package dao
-
-import (
-	"context"
-
-	"gitlab.wanzhuangkj.com/tush/xpkg/database"
-	"gitlab.wanzhuangkj.com/tush/xpkg/pkg/eventbus"
-	"gitlab.wanzhuangkj.com/tush/xpkg/xcron/cache"
-)
-
-var (
-	Schema string
-)
-
-func Init(schema string) {
-	Schema = schema
-	_ = eventbus.Eb.Subscribe(eventbus.TopicCacheInitFinish, func(ctx context.Context) {
-		CronJobDao.ODao.Cache = cache.NewCronJobCache(database.GetCacheType())
-	})
-}

xcron/dao/init.go → xcron/init.go

-package dao
+package xcron
 
 import (
 	"context"
@@ -6,16 +6,32 @@ import (
 	"gitlab.wanzhuangkj.com/tush/xpkg/config"
 	"gitlab.wanzhuangkj.com/tush/xpkg/database"
 	"gitlab.wanzhuangkj.com/tush/xpkg/pkg/eventbus"
+	"gitlab.wanzhuangkj.com/tush/xpkg/xcron/cache"
+	"gitlab.wanzhuangkj.com/tush/xpkg/xcron/dao"
 )
 
 func init() {
 	_ = eventbus.Eb.Subscribe(eventbus.TopicParseConfFinish, func(ctx context.Context) {
+		cronCfg := config.GetCron()
+		if !cronCfg.Enable {
+			return
+		}
 		schema := ""
 		config.Read(func(c *config.Config) {
 			schema = c.Cron.Schema
 		})
-		CronJobDao.ODao.XDB = &database.XDB{Schema: schema}
-		CronJobLogDao.ODao.XDB = &database.XDB{Schema: schema}
-		CronJobRecordDao.ODao.XDB = &database.XDB{Schema: schema}
+		dao.CronJobDao.ODao.XDB = &database.XDB{Schema: schema}
+		dao.CronJobLogDao.ODao.XDB = &database.XDB{Schema: schema}
+		dao.CronJobRecordDao.ODao.XDB = &database.XDB{Schema: schema}
+	})
+	_ = eventbus.Eb.Subscribe(eventbus.TopicCacheInitFinish, func(ctx context.Context) {
+		cronCfg := config.GetCron()
+		if !cronCfg.Enable {
+			return
+		}
+		dao.CronJobDao.ODao.Cache = cache.NewCronJobCache(database.GetCacheType())
+	})
+	_ = eventbus.Eb.Subscribe(eventbus.TopicCoreInitFinish, func(ctx context.Context) {
+		initCron()
 	})
 }