From dde277c14d1410c8acce1df25b754a1d1e936d3a Mon Sep 17 00:00:00 2001 From: huang <1724659546@qq.com> Date: Wed, 17 Sep 2025 16:55:56 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0task=E5=B7=A5=E5=8E=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/app/service/task/delay_task.go | 13 ++--- .../app/service/task/plan_analysis_task.go | 9 ---- internal/app/service/task/task.go | 17 +++++++ internal/infra/models/plan.go | 3 +- internal/infra/task/plan_analysis_task.go | 29 +++++++++++ internal/infra/task/scheduler.go | 51 +++++++++++-------- internal/infra/task/task.go | 7 ++- 7 files changed, 89 insertions(+), 40 deletions(-) delete mode 100644 internal/app/service/task/plan_analysis_task.go create mode 100644 internal/app/service/task/task.go create mode 100644 internal/infra/task/plan_analysis_task.go diff --git a/internal/app/service/task/delay_task.go b/internal/app/service/task/delay_task.go index d6be350..c4960ed 100644 --- a/internal/app/service/task/delay_task.go +++ b/internal/app/service/task/delay_task.go @@ -20,14 +20,6 @@ type DelayTask struct { logger *logs.Logger } -// NewDelayTask 创建一个新的 DelayTask 实例 -func NewDelayTask(logger *logs.Logger, executionTask *models.TaskExecutionLog) *DelayTask { - return &DelayTask{ - executionTask: executionTask, - logger: logger, - } -} - // Execute 执行延迟任务，等待指定的时间 func (d *DelayTask) Execute() error { d.logger.Infof("任务 %v: 开始延迟 %v...", d.executionTask.TaskID, d.duration) @@ -36,7 +28,10 @@ func (d *DelayTask) Execute() error { return nil } -func (d *DelayTask) ParseParams() error { +func (d *DelayTask) ParseParams(logger *logs.Logger, executionTask *models.TaskExecutionLog) error { + d.logger = logger + d.executionTask = executionTask + if d.executionTask.Task.Parameters == nil { d.logger.Errorf("任务 %v: 缺少参数", d.executionTask.TaskID) return fmt.Errorf("任务 %v: 参数不全", d.executionTask.TaskID) diff --git a/internal/app/service/task/plan_analysis_task.go b/internal/app/service/task/plan_analysis_task.go deleted file mode 100644 index d3a8dfa..0000000 --- a/internal/app/service/task/plan_analysis_task.go +++ /dev/null @@ -1,9 +0,0 @@ -package task - -// PlanAnalysisTask 用于在任务执行队列中触发一个plan的执行 -// 该任务会解析plan生成扁平化的待执行任务表, 并将任务列表插入任务执行队列 -// 该任务会预写入plan所有待执行任务的执行日志 -// 每个plan执行完毕时 或 创建plan时 都应该重新创建一个 PlanAnalysisTask 以便触发下次plan执行 -// 更新plan后应当更新对应 PlanAnalysisTask -type PlanAnalysisTask struct { -} diff --git a/internal/app/service/task/task.go b/internal/app/service/task/task.go new file mode 100644 index 0000000..b66a289 --- /dev/null +++ b/internal/app/service/task/task.go @@ -0,0 +1,17 @@ +package task + +import ( + "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/task" +) + +// TaskFactory 是一个任务组装工厂, 可以根据Task类型获取到对应的初始化函数 +var TaskFactory = func(tt models.TaskType) task.Task { + switch tt { + case models.TaskTypeWaiting: + return &DelayTask{} + default: + // 出现位置任务类型说明业务逻辑出现重大问题, 一个异常任务被创建了出来 + panic("发现未知任务类型") + } +} diff --git a/internal/infra/models/plan.go b/internal/infra/models/plan.go index 6f9abbc..e26aca2 100644 --- a/internal/infra/models/plan.go +++ b/internal/infra/models/plan.go @@ -29,7 +29,8 @@ const ( type TaskType string const ( - TaskTypeWaiting TaskType = "waiting" // 等待任务 + TaskPlanAnalysis TaskType = "plan_analysis" // 解析Plan的Task列表并添加到待执行队列的特殊任务 + TaskTypeWaiting TaskType = "waiting" // 等待任务 ) // Plan 代表系统中的一个计划，可以包含子计划或任务 diff --git a/internal/infra/task/plan_analysis_task.go b/internal/infra/task/plan_analysis_task.go new file mode 100644 index 0000000..e48e41f --- /dev/null +++ b/internal/infra/task/plan_analysis_task.go @@ -0,0 +1,29 @@ +package task + +import ( + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" +) + +// PlanAnalysisTask 用于在任务执行队列中触发一个plan的执行 +// 该任务会解析plan生成扁平化的待执行任务表, 并将任务列表插入任务执行队列 +// 该任务会预写入plan所有待执行任务的执行日志 +// 每个plan执行完毕时 或 创建plan时 都应该重新创建一个 PlanAnalysisTask 以便触发下次plan执行 +// 更新plan后应当更新对应 PlanAnalysisTask +type PlanAnalysisTask struct { +} + +func (p *PlanAnalysisTask) Execute() error { + //TODO implement me + panic("implement me") +} + +func (p *PlanAnalysisTask) ParseParams(logger *logs.Logger, claimedLog *models.TaskExecutionLog) error { + //TODO implement me + panic("implement me") +} + +func (p *PlanAnalysisTask) OnFailure(executeErr error) { + //TODO implement me + panic("implement me") +} diff --git a/internal/infra/task/scheduler.go b/internal/infra/task/scheduler.go index 387e02c..cfd610e 100644 --- a/internal/infra/task/scheduler.go +++ b/internal/infra/task/scheduler.go @@ -6,17 +6,13 @@ import ( "sync" "time" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" "git.huangwc.com/pig/pig-farm-controller/internal/infra/repository" "github.com/panjf2000/ants/v2" "gorm.io/gorm" ) -// Logger 定义了调度器期望的日志接口，方便替换为项目中的日志组件 -type Logger interface { - Printf(format string, v ...interface{}) -} - // ProgressTracker 在内存中跟踪计划的执行状态，包括进度和执行锁 type ProgressTracker struct { mu sync.Mutex @@ -83,11 +79,12 @@ func (t *ProgressTracker) GetRunningPlanIDs() []uint { // Scheduler 是核心的、持久化的任务调度器 type Scheduler struct { - logger Logger + logger *logs.Logger pollingInterval time.Duration workers int pendingTaskRepo repository.PendingTaskRepository progressTracker *ProgressTracker + taskFactory func(taskType models.TaskType) Task // 调度器需要注入一个任务工厂，用于创建任务实例 pool *ants.Pool // 使用 ants 协程池来管理并发 wg sync.WaitGroup @@ -96,7 +93,7 @@ type Scheduler struct { } // NewScheduler 创建一个新的调度器实例 -func NewScheduler(pendingTaskRepo repository.PendingTaskRepository, logger Logger, interval time.Duration, numWorkers int) *Scheduler { +func NewScheduler(pendingTaskRepo repository.PendingTaskRepository, taskFactory func(taskType models.TaskType) Task, logger *logs.Logger, interval time.Duration, numWorkers int) *Scheduler { ctx, cancel := context.WithCancel(context.Background()) return &Scheduler{ pendingTaskRepo: pendingTaskRepo, @@ -104,6 +101,7 @@ func NewScheduler(pendingTaskRepo repository.PendingTaskRepository, logger Logge pollingInterval: interval, workers: numWorkers, progressTracker: NewProgressTracker(), + taskFactory: taskFactory, ctx: ctx, cancel: cancel, } @@ -111,9 +109,9 @@ func NewScheduler(pendingTaskRepo repository.PendingTaskRepository, logger Logge // Start 启动调度器，包括初始化协程池和启动主轮询循环 func (s *Scheduler) Start() { - s.logger.Printf("任务调度器正在启动，工作协程数: %d...", s.workers) + s.logger.Warnf("任务调度器正在启动，工作协程数: %d...", s.workers) pool, err := ants.NewPool(s.workers, ants.WithPanicHandler(func(err interface{}) { - s.logger.Printf("[严重] 任务执行时发生 panic: %v", err) + s.logger.Errorf("[严重] 任务执行时发生 panic: %v", err) })) if err != nil { panic("初始化协程池失败: " + err.Error()) @@ -122,16 +120,16 @@ func (s *Scheduler) Start() { s.wg.Add(1) go s.run() - s.logger.Printf("任务调度器已成功启动") + s.logger.Warnf("任务调度器已成功启动") } // Stop 优雅地停止调度器 func (s *Scheduler) Stop() { - s.logger.Printf("正在停止任务调度器...") + s.logger.Warnf("正在停止任务调度器...") s.cancel() // 1. 发出取消信号，停止主循环 s.wg.Wait() // 2. 等待主循环完成 s.pool.Release() // 3. 释放 ants 池 (等待所有已提交的任务执行完毕) - s.logger.Printf("任务调度器已安全停止") + s.logger.Warnf("任务调度器已安全停止") } // run 是主轮询循环，负责从数据库认领任务并提交到协程池 @@ -157,7 +155,7 @@ func (s *Scheduler) claimAndSubmit() { claimedLog, pendingTask, err := s.pendingTaskRepo.ClaimNextAvailableTask(runningPlanIDs) if err != nil { if !errors.Is(err, gorm.ErrRecordNotFound) { - s.logger.Printf("认领任务时发生错误: %v", err) + s.logger.Errorf("认领任务时发生错误: %v", err) } // gorm.ErrRecordNotFound 说明没任务要执行 return @@ -171,7 +169,7 @@ func (s *Scheduler) claimAndSubmit() { s.processTask(claimedLog) }) if err != nil { - s.logger.Printf("向协程池提交任务失败: %v", err) + s.logger.Errorf("向协程池提交任务失败: %v", err) // 提交失败，必须释放刚刚获取的锁 s.progressTracker.Unlock(claimedLog.PlanExecutionLogID) // 同样需要将任务安全放回 @@ -185,7 +183,7 @@ func (s *Scheduler) claimAndSubmit() { // handleRequeue 同步地、安全地将一个无法立即执行的任务放回队列。 func (s *Scheduler) handleRequeue(planExecutionLogID uint, taskToRequeue *models.PendingTask) { - s.logger.Printf("计划 %d 正在执行，任务 %d (TaskID: %d) 将等待并重新入队...", planExecutionLogID, taskToRequeue.ID, taskToRequeue.TaskID) + s.logger.Warnf("计划 %d 正在执行，任务 %d (TaskID: %d) 将等待并重新入队...", planExecutionLogID, taskToRequeue.ID, taskToRequeue.TaskID) // 1. 阻塞式地等待，直到可以获取到该计划的锁。 s.progressTracker.Lock(planExecutionLogID) @@ -193,17 +191,30 @@ func (s *Scheduler) handleRequeue(planExecutionLogID uint, taskToRequeue *models // 2. 在持有锁的情况下，将任务安全地放回队列。 if err := s.pendingTaskRepo.RequeueTask(taskToRequeue); err != nil { - s.logger.Printf("[严重] 任务重新入队失败, 原始PendingTaskID: %d, 错误: %v", taskToRequeue.ID, err) + s.logger.Errorf("[严重] 任务重新入队失败, 原始PendingTaskID: %d, 错误: %v", taskToRequeue.ID, err) return } - s.logger.Printf("任务 (原始ID: %d) 已成功重新入队，并已释放计划 %d 的锁。", taskToRequeue.ID, planExecutionLogID) + s.logger.Warnf("任务 (原始ID: %d) 已成功重新入队，并已释放计划 %d 的锁。", taskToRequeue.ID, planExecutionLogID) } // processTask 处理单个任务的逻辑 (当前为占位符) func (s *Scheduler) processTask(claimedLog *models.TaskExecutionLog) { - s.logger.Printf("开始处理任务, 日志ID: %d, 任务ID: %d, 任务名称: %s", + s.logger.Warnf("开始处理任务, 日志ID: %d, 任务ID: %d, 任务名称: %s", claimedLog.ID, claimedLog.TaskID, claimedLog.Task.Name) - time.Sleep(2 * time.Second) // 模拟任务执行 - s.logger.Printf("完成任务, 日志ID: %d", claimedLog.ID) + + task := s.taskFactory(claimedLog.Task.Type) + if err := task.ParseParams(s.logger, claimedLog); err != nil { + s.logger.Errorf("[严重] 任务参数解析失败, 日志ID: %d, 错误: %v", claimedLog.ID, err) + return + } + + if err := task.Execute(); err != nil { + s.logger.Errorf("[严重] 任务执行失败, 日志ID: %d, 错误: %v", claimedLog.ID, err) + + task.OnFailure(err) + return + } + + s.logger.Warnf("完成任务, 日志ID: %d", claimedLog.ID) } diff --git a/internal/infra/task/task.go b/internal/infra/task/task.go index 3455e20..411c715 100644 --- a/internal/infra/task/task.go +++ b/internal/infra/task/task.go @@ -1,5 +1,10 @@ package task +import ( + "git.huangwc.com/pig/pig-farm-controller/internal/infra/logs" + "git.huangwc.com/pig/pig-farm-controller/internal/infra/models" +) + // Task 定义了所有可被调度器执行的任务必须实现的接口。 type Task interface { // Execute 是任务的核心执行逻辑。 @@ -9,7 +14,7 @@ type Task interface { Execute() error // ParseParams 解析参数 - ParseParams() error + ParseParams(logger *logs.Logger, claimedLog *models.TaskExecutionLog) error // OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。 // log: 任务执行的上下文。