FranklinWH-Toolbox/service/task/core/taskCore.go

package task_core

import (
	task_pg "fwh-toolbox-server/task/plugin"

	"github.com/hibiken/asynq"
	"go.uber.org/zap"
)

// type TaskCore interface {
// 	CreateTask() *asynq.Client
// }

type TaskCore struct {
	redisOpt asynq.RedisClientOpt
	client   *asynq.Client

	Scheduler *task_pg.SchedulerManager
	Inspector *task_pg.InspectorManager
}

type TaskConfig struct {
	Name   string // 任务名
	SvrCfg asynq.Config
	Mux    *asynq.ServeMux
	Agg    *task_pg.AggregatorManager // 可选
}

// 创建TaskCore
func NewTaskCore(redisOpt asynq.RedisClientOpt) *TaskCore {
	return &TaskCore{
		redisOpt:  redisOpt,
		client:    asynq.NewClient(redisOpt),
		Scheduler: task_pg.NewSchedulerManager(redisOpt),
		Inspector: task_pg.NewInspectorManager(redisOpt),
	}
}

// 创建任务
func (tc *TaskCore) CreateTask(config TaskConfig) *asynq.Client {
	if config.Mux != nil {
		svr := asynq.NewServer(tc.redisOpt, config.SvrCfg)

		go func() {
			err := svr.Run(config.Mux)
			if err != nil {
				zap.L().Error("could not run server", zap.String("err", err.Error()), zap.String("taskName", config.Name))
				return
			}
			zap.L().Warn("asynq server shutdown", zap.String("taskName", config.Name))
		}()
	} else {
		zap.L().Warn("asynq not server", zap.String("taskName", config.Name))
	}

	if tc.client == nil {
		tc.client = asynq.NewClient(tc.redisOpt)
	}
	return tc.client
}