任务调度器关于任务执行部分实现(没测没检查, 但应该实现完了)

TODO-List

@@ -1,9 +1,10 @@
 // TODO 列表
 
 // TODO 可以实现的问题
-plan执行到一半时如果用户删掉里面的task, 后续调度器执行task时可能会找不到这个任务的细节
-1. 可以用TimescaleDB代替PGSQL, 优化传感器数据存储性能
-2. 系统启动时应该检查一遍执行历史库, 将所有显示为执行中的任务都修正为执行失败并报错
+1. plan执行到一半时如果用户删掉里面的task, 后续调度器执行task时可能会找不到这个任务的细节
+2. 创建/更新plan的地方需要创建/更新触发器Task, 调度器那里不再创建新的Task, 只创建PendingTask
+3. 可以用TimescaleDB代替PGSQL, 优化传感器数据存储性能
+4. 系统启动时应该检查一遍执行历史库, 将所有显示为执行中的任务都修正为执行失败并报错
 
 // TODO 暂时实现不了的问题
 1. 目前设备都只对应一个地址, 但实际上如电磁两位五通阀等设备是需要用两个开关控制的

go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/go-openapi/validate v0.24.0
 	github.com/golang-jwt/jwt/v5 v5.3.0
 	github.com/panjf2000/ants/v2 v2.11.3
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/stretchr/testify v1.11.1
 	github.com/swaggo/files v1.0.1
 	github.com/swaggo/gin-swagger v1.6.1

go.sum

@@ -123,6 +123,8 @@ github.com/pelletier/go-toml/v2 v2.2.2 h1:aYUidT7k73Pcl9nb2gScu7NSrKCSHIDE89b3+6
 github.com/pelletier/go-toml/v2 v2.2.2/go.mod h1:1t835xjRzz80PqgE6HHgN2JOsmgYu/h4qDAS4n929Rs=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=

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)

internal/app/service/task/plan_analysis_task.go

@@ -1,9 +0,0 @@
-package task
-
-// PlanAnalysisTask 用于在任务执行队列中触发一个plan的执行
-// 该任务会解析plan生成扁平化的待执行任务表, 并将任务列表插入任务执行队列
-// 该任务会预写入plan所有待执行任务的执行日志
-// 每个plan执行完毕时 或 创建plan时 都应该重新创建一个 PlanAnalysisTask 以便触发下次plan执行
-// 更新plan后应当更新对应 PlanAnalysisTask
-type PlanAnalysisTask struct {
-}

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("发现未知任务类型")
+	}
+}

internal/infra/models/plan.go

@@ -29,9 +29,15 @@ const (
 type TaskType string
 
 const (
+	TaskPlanAnalysis TaskType = "plan_analysis" // 解析Plan的Task列表并添加到待执行队列的特殊任务
 	TaskTypeWaiting  TaskType = "waiting"       // 等待任务
 )
 
+const (
+	// 这个参数是 TaskPlanAnalysis 类型的 Task Parameters 中用于记录plan_id的字段的key
+	ParamsPlanID = "plan_id"
+)
+
 // Plan 代表系统中的一个计划，可以包含子计划或任务
 type Plan struct {
 	gorm.Model

internal/infra/repository/execution_log_repository.go

@@ -12,6 +12,7 @@ type ExecutionLogRepository interface {
 	UpdatePlanExecutionLog(log *models.PlanExecutionLog) error
 	CreateTaskExecutionLogsInBatch(logs []*models.TaskExecutionLog) error
 	UpdateTaskExecutionLog(log *models.TaskExecutionLog) error
+	FindTaskExecutionLogByID(id uint) (*models.TaskExecutionLog, error)
 }
 
 // executionLogRepository 是使用 GORM 的具体实现。
@@ -30,11 +31,11 @@ func (r *executionLogRepository) CreatePlanExecutionLog(log *models.PlanExecutio
 	return r.db.Create(log).Error
 }
 
-// UpdatePlanExecutionLog 使用 Save 方法全量更新一个计划执行日志。
-// GORM 的 Save 会自动根据主键是否存在来决定是执行 UPDATE 还是 INSERT。
+// UpdatePlanExecutionLog 使用 Updates 方法更新一个计划执行日志。
+// GORM 的 Updates 传入 struct 时，只会更新非零值字段。
 // 在这里，我们期望传入的对象一定包含一个有效的 ID。
 func (r *executionLogRepository) UpdatePlanExecutionLog(log *models.PlanExecutionLog) error {
-	return r.db.Save(log).Error
+	return r.db.Updates(log).Error
 }
 
 // CreateTaskExecutionLogsInBatch 在一次数据库调用中创建多个任务执行日志条目。
@@ -44,8 +45,21 @@ func (r *executionLogRepository) CreateTaskExecutionLogsInBatch(logs []*models.T
 	return r.db.Create(&logs).Error
 }
 
-// UpdateTaskExecutionLog 使用 Save 方法全量更新一个任务执行日志。
+// UpdateTaskExecutionLog 使用 Updates 方法更新一个任务执行日志。
+// GORM 的 Updates 传入 struct 时，只会更新非零值字段。
 // 这种方式代码更直观，上层服务可以直接修改模型对象后进行保存。
 func (r *executionLogRepository) UpdateTaskExecutionLog(log *models.TaskExecutionLog) error {
-	return r.db.Save(log).Error
+	return r.db.Updates(log).Error
+}
+
+// FindTaskExecutionLogByID 根据 ID 查找单个任务执行日志。
+// 它会预加载关联的 Task 信息。
+func (r *executionLogRepository) FindTaskExecutionLogByID(id uint) (*models.TaskExecutionLog, error) {
+	var log models.TaskExecutionLog
+	// 使用 Preload("Task") 来确保关联的任务信息被一并加载
+	err := r.db.Preload("Task").First(&log, id).Error
+	if err != nil {
+		return nil, err
+	}
+	return &log, nil
 }

internal/infra/repository/plan_repository.go

@@ -34,6 +34,12 @@ type PlanRepository interface {
 	UpdatePlan(plan *models.Plan) error
 	// DeletePlan 根据ID删除计划，同时删除其关联的任务（非子任务）或子计划关联
 	DeletePlan(id uint) error
+	// FlattenPlanTasks 递归展开计划，返回按执行顺序排列的所有任务列表
+	FlattenPlanTasks(planID uint) ([]models.Task, error)
+	// DeleteTask 根据ID删除任务
+	DeleteTask(id int) error
+	// FindPlanAnalysisTaskByParamsPlanID 根据Parameters中的ParamsPlanID字段值查找TaskPlanAnalysis类型的Task
+	FindPlanAnalysisTaskByParamsPlanID(paramsPlanID uint) (*models.Task, error)
 }
 
 // gormPlanRepository 是 PlanRepository 的 GORM 实现
@@ -288,7 +294,7 @@ func (r *gormPlanRepository) reconcileTasks(tx *gorm.DB, plan *models.Plan) erro
 		return err
 	}
 
-	existingTaskMap := make(map[uint]bool)
+	existingTaskMap := make(map[int]bool)
 	for _, task := range existingTasks {
 		existingTaskMap[task.ID] = true
 	}
@@ -308,7 +314,7 @@ func (r *gormPlanRepository) reconcileTasks(tx *gorm.DB, plan *models.Plan) erro
 		}
 	}
 
-	var tasksToDelete []uint
+	var tasksToDelete []int
 	for id := range existingTaskMap {
 		tasksToDelete = append(tasksToDelete, id)
 	}
@@ -401,3 +407,116 @@ func (r *gormPlanRepository) DeletePlan(id uint) error {
 		return nil
 	})
 }
+
+// FlattenPlanTasks 递归展开计划，返回按执行顺序排列的所有任务列表
+func (r *gormPlanRepository) FlattenPlanTasks(planID uint) ([]models.Task, error) {
+	plan, err := r.GetPlanByID(planID)
+	if err != nil {
+		return nil, fmt.Errorf("获取计划(ID: %d)失败: %w", planID, err)
+	}
+
+	return r.flattenPlanTasksRecursive(plan)
+}
+
+// flattenPlanTasksRecursive 递归展开计划的内部实现
+func (r *gormPlanRepository) flattenPlanTasksRecursive(plan *models.Plan) ([]models.Task, error) {
+	var tasks []models.Task
+
+	switch plan.ContentType {
+	case models.PlanContentTypeTasks:
+		// 如果计划直接包含任务，直接返回这些任务
+		// 由于GetPlanByID已经预加载并排序了任务，这里直接使用即可
+		tasks = append(tasks, plan.Tasks...)
+
+	case models.PlanContentTypeSubPlans:
+		// 如果计划包含子计划，则递归处理每个子计划
+		for _, subPlan := range plan.SubPlans {
+			// 获取子计划的任务列表
+			var subTasks []models.Task
+			var err error
+
+			// 确保子计划已经被加载
+			if subPlan.ChildPlan != nil {
+				subTasks, err = r.flattenPlanTasksRecursive(subPlan.ChildPlan)
+			} else {
+				// 如果子计划未加载，则从数据库获取并递归展开
+				subTasks, err = r.FlattenPlanTasks(subPlan.ChildPlanID)
+			}
+
+			if err != nil {
+				return nil, fmt.Errorf("展开子计划(ID: %d)失败: %w", subPlan.ChildPlanID, err)
+			}
+
+			// 将子计划的任务添加到结果中
+			tasks = append(tasks, subTasks...)
+		}
+	default:
+		return nil, fmt.Errorf("未知的计划内容类型: %v", plan.ContentType)
+	}
+
+	return tasks, nil
+}
+
+// DeleteTask 根据ID删除任务
+func (r *gormPlanRepository) DeleteTask(id int) error {
+	// 使用事务确保操作的原子性
+	return r.db.Transaction(func(tx *gorm.DB) error {
+		// 1. 检查是否有待执行任务引用了这个任务
+		var pendingTaskCount int64
+		if err := tx.Model(&models.PendingTask{}).Where("task_id = ?", id).Count(&pendingTaskCount).Error; err != nil {
+			return fmt.Errorf("检查待执行任务时出错: %w", err)
+		}
+
+		// 如果有待执行任务引用该任务，不能删除
+		if pendingTaskCount > 0 {
+			return fmt.Errorf("无法删除任务(ID: %d)，因为存在 %d 条待执行任务引用该任务", id, pendingTaskCount)
+		}
+
+		// 2. 检查是否有计划仍在使用这个任务
+		var planCount int64
+		if err := tx.Model(&models.Plan{}).Joins("JOIN tasks ON plans.id = tasks.plan_id").Where("tasks.id = ?", id).Count(&planCount).Error; err != nil {
+			return fmt.Errorf("检查计划引用任务时出错: %w", err)
+		}
+
+		// 如果有计划在使用该任务，不能删除
+		if planCount > 0 {
+			return fmt.Errorf("无法删除任务(ID: %d)，因为存在 %d 个计划仍在使用该任务", id, planCount)
+		}
+
+		// 3. 执行删除操作
+		result := tx.Delete(&models.Task{}, id)
+		if result.Error != nil {
+			return fmt.Errorf("删除任务失败: %w", result.Error)
+		}
+
+		// 检查是否实际删除了记录
+		if result.RowsAffected == 0 {
+			return gorm.ErrRecordNotFound
+		}
+
+		return nil
+	})
+}
+
+// FindPlanAnalysisTaskByParamsPlanID 根据Parameters中的ParamsPlanID字段值查找TaskPlanAnalysis类型的Task
+func (r *gormPlanRepository) FindPlanAnalysisTaskByParamsPlanID(paramsPlanID uint) (*models.Task, error) {
+	var task models.Task
+
+	// 构造JSON查询条件，查找Parameters中包含指定ParamsPlanID且Type为TaskPlanAnalysis的任务
+	// TODO 在JSON字段中查找特定键值的语法取决于数据库类型，这里使用PostgreSQL的语法
+	// TODO 如果使用的是MySQL，则需要相应调整查询条件
+	result := r.db.Where(
+		"type = ? AND parameters->>'plan_id' = ?",
+		models.TaskPlanAnalysis,
+		fmt.Sprintf("%d", paramsPlanID),
+	).First(&task)
+
+	if result.Error != nil {
+		if errors.Is(result.Error, gorm.ErrRecordNotFound) {
+			return nil, fmt.Errorf("未找到Parameters.PlanID为%d的TaskPlanAnalysis类型任务", paramsPlanID)
+		}
+		return nil, fmt.Errorf("查找任务时出错: %w", result.Error)
+	}
+
+	return &task, nil
+}

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")
+}

internal/infra/task/scheduler.go

@@ -6,17 +6,14 @@ 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"
+	"git.huangwc.com/pig/pig-farm-controller/internal/infra/utils"
 	"github.com/panjf2000/ants/v2"
 	"gorm.io/gorm"
 )
 
-// Logger 定义了调度器期望的日志接口，方便替换为项目中的日志组件
-type Logger interface {
-	Printf(format string, v ...interface{})
-}
-
 // ProgressTracker 在内存中跟踪计划的执行状态，包括进度和执行锁
 type ProgressTracker struct {
 	mu             sync.Mutex
@@ -37,6 +34,29 @@ func NewProgressTracker() *ProgressTracker {
 	return t
 }
 
+// AddNewPlan 添加一个新的计划，并初始化进度跟踪器
+func (t *ProgressTracker) AddNewPlan(planLogID uint, totalTasks int) {
+	t.mu.Lock()
+	t.totalTasks[planLogID] = totalTasks
+	t.completedTasks[planLogID] = 0
+	t.mu.Unlock()
+}
+
+// CompletedTask 通知计数器一个任务被完成
+func (t *ProgressTracker) CompletedTask(planLogID uint) {
+	t.mu.Lock()
+	t.completedTasks[planLogID]++
+	t.mu.Unlock()
+}
+
+// IsPlanOver 检查计划是否完成
+func (t *ProgressTracker) IsPlanOver(planLogID uint) bool {
+	t.mu.Lock()
+	defer t.mu.Unlock()
+
+	return t.completedTasks[planLogID] >= t.totalTasks[planLogID]
+}
+
 // TryLock (非阻塞) 尝试锁定一个计划。如果计划未被锁定，则锁定并返回 true。
 func (t *ProgressTracker) TryLock(planLogID uint) bool {
 	t.mu.Lock()
@@ -83,11 +103,14 @@ func (t *ProgressTracker) GetRunningPlanIDs() []uint {
 
 // Scheduler 是核心的、持久化的任务调度器
 type Scheduler struct {
-	logger          Logger
+	logger           *logs.Logger
 	pollingInterval  time.Duration
 	workers          int
 	pendingTaskRepo  repository.PendingTaskRepository
+	executionLogRepo repository.ExecutionLogRepository
+	planRepo         repository.PlanRepository
 	progressTracker  *ProgressTracker
+	taskFactory      func(taskType models.TaskType) Task // 调度器需要注入一个任务工厂，用于创建任务实例
 
 	pool   *ants.Pool // 使用 ants 协程池来管理并发
 	wg     sync.WaitGroup
@@ -96,14 +119,25 @@ type Scheduler struct {
 }
 
 // NewScheduler 创建一个新的调度器实例
-func NewScheduler(pendingTaskRepo repository.PendingTaskRepository, logger Logger, interval time.Duration, numWorkers int) *Scheduler {
+func NewScheduler(
+	pendingTaskRepo repository.PendingTaskRepository,
+	executionLogRepo repository.ExecutionLogRepository,
+	planRepo repository.PlanRepository,
+	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,
+		executionLogRepo: executionLogRepo,
+		planRepo:         planRepo,
 		logger:           logger,
 		pollingInterval:  interval,
 		workers:          numWorkers,
 		progressTracker:  NewProgressTracker(),
+		taskFactory:      taskFactory,
 		ctx:              ctx,
 		cancel:           cancel,
 	}
@@ -111,9 +145,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 +156,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 +191,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 +205,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 +219,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 +227,186 @@ 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)
+
+	claimedLog.StartedAt = time.Now()
+	claimedLog.Status = models.ExecutionStatusCompleted // 先乐观假定任务成功, 后续失败了再改
+	defer s.updateTaskExecutionLogStatus(claimedLog)
+
+	// 执行任务
+	err := s.runTask(claimedLog)
+	if err != nil {
+		claimedLog.Status = models.ExecutionStatusFailed
+		claimedLog.Output = err.Error()
+		return
+	}
+	s.logger.Warnf("完成任务, 日志ID: %d", claimedLog.ID)
+
+	// 任务计数器校验, Plan的任务全部执行完成后需要插入一个新的PlanAnalysisTask用于触发下一次Plan的执行
+	if s.progressTracker.IsPlanOver(claimedLog.PlanExecutionLogID) {
+		err = s.createNewAnalysisPlanTask(claimedLog.Task.PlanID)
+		if err != nil {
+			s.logger.Errorf("[严重] 创建计划分析任务失败, 当前Plan(%v)将无法进行下次触发, 错误: %v", claimedLog.Task.PlanID, err)
+		}
+	}
+
+}
+
+// runTask 用于执行具体任务
+func (s *Scheduler) runTask(claimedLog *models.TaskExecutionLog) error {
+	// 这是个特殊任务, 用于解析Plan并将解析出的任务队列添加到待执行队列中
+	if claimedLog.Task.Type == models.TaskPlanAnalysis {
+		// 解析plan
+		err := s.analysisPlan(claimedLog)
+		if err != nil {
+			// TODO 这里要处理一下, 比如再插一个新的触发器回去
+			s.logger.Errorf("[严重] 计划解析失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+			return err
+		}
+
+	} else {
+		// 执行普通任务
+		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 err
+		}
+
+		if err := task.Execute(); err != nil {
+
+			s.logger.Errorf("[严重] 任务执行失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+
+			task.OnFailure(err)
+			return err
+		}
+
+	}
+	return nil
+}
+
+// analysisPlan 解析Plan并将解析出的Task列表插入待执行队列中
+func (s *Scheduler) analysisPlan(claimedLog *models.TaskExecutionLog) error {
+	// 创建Plan执行记录
+	planLog := &models.PlanExecutionLog{
+		PlanID:    claimedLog.Task.PlanID,
+		Status:    models.ExecutionStatusStarted,
+		StartedAt: time.Now(),
+	}
+	if err := s.executionLogRepo.CreatePlanExecutionLog(planLog); err != nil {
+		s.logger.Errorf("[严重] 创建计划执行日志失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+		return err
+	}
+
+	// 解析出Task列表
+	tasks, err := s.planRepo.FlattenPlanTasks(claimedLog.Task.PlanID)
+	if err != nil {
+		s.logger.Errorf("[严重] 解析计划失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+		return err
+	}
+
+	// 写入执行历史
+	taskLogs := make([]*models.TaskExecutionLog, len(tasks))
+	for _, task := range tasks {
+		taskLogs = append(taskLogs, &models.TaskExecutionLog{
+			PlanExecutionLogID: planLog.ID,
+			TaskID:             task.ID,
+			Status:             models.ExecutionStatusWaiting,
+		})
+
+	}
+	err = s.executionLogRepo.CreateTaskExecutionLogsInBatch(taskLogs)
+	if err != nil {
+		s.logger.Errorf("[严重] 写入执行历史, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+		return err
+	}
+
+	// 写入待执行队列
+	pendingTasks := make([]*models.PendingTask, len(tasks))
+	for i, task := range tasks {
+		pendingTasks = append(pendingTasks, &models.PendingTask{
+			TaskID:             task.ID,
+			TaskExecutionLogID: pendingTasks[i].ID,
+
+			// 待执行队列是通过任务触发时间排序的, 且只要在调度器获取的时间点之前的都可以被触发
+			ExecuteAt: time.Now().Add(time.Duration(i) * time.Second),
+		})
+	}
+	err = s.pendingTaskRepo.CreatePendingTasksInBatch(pendingTasks)
+	if err != nil {
+		s.logger.Errorf("[严重] 写入待执行队列, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+		return err
+	}
+
+	// 将Task列表加入待执行队列中
+	s.progressTracker.AddNewPlan(claimedLog.PlanExecutionLogID, len(tasks))
+
+	return nil
+}
+
+// updateTaskExecutionLogStatus 修改任务历史中的执行状态
+func (s *Scheduler) updateTaskExecutionLogStatus(claimedLog *models.TaskExecutionLog) error {
+	claimedLog.EndedAt = time.Now()
+
+	if err := s.executionLogRepo.UpdateTaskExecutionLog(claimedLog); err != nil {
+		s.logger.Errorf("[严重] 更新任务执行日志失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
+		return err
+	}
+
+	return nil
+}
+
+// createNewAnalysisPlanTask 创建一个新的Plan解析任务用于下次触发plan执行
+func (s *Scheduler) createNewAnalysisPlanTask(planID uint) error {
+	// 获取计划信息
+	plan, err := s.planRepo.GetBasicPlanByID(planID)
+	if err != nil {
+		s.logger.Errorf("[严重] 获取计划失败, 错误: %v", err)
+		return err
+	}
+
+	// 获取触发任务
+	task, err := s.planRepo.FindPlanAnalysisTaskByParamsPlanID(planID)
+	if err != nil {
+		s.logger.Errorf("[严重] 获取计划解析任务失败, 错误: %v", err)
+		return err
+	}
+
+	// 写入执行日志
+	taskLog := &models.TaskExecutionLog{
+		TaskID: task.ID,
+		Status: models.ExecutionStatusWaiting,
+	}
+	if err := s.executionLogRepo.CreateTaskExecutionLogsInBatch([]*models.TaskExecutionLog{taskLog}); err != nil {
+		s.logger.Errorf("[严重] 创建任务执行日志失败, 错误: %v", err)
+		return err
+	}
+
+	// 写入待执行队列
+	next, err := utils.GetNextCronTime(plan.CronExpression)
+	if err != nil {
+		s.logger.Errorf("[严重] 执行时间解析失败, 错误: %v", err)
+		return err
+	}
+	pendingTask := &models.PendingTask{
+		TaskID:             task.ID,
+		ExecuteAt:          next,
+		TaskExecutionLogID: taskLog.ID,
+	}
+	err = s.pendingTaskRepo.CreatePendingTasksInBatch([]*models.PendingTask{pendingTask})
+	if err != nil {
+		s.logger.Errorf("[严重] 创建待执行任务失败, 错误: %v", err)
+		return err
+	}
+	return nil
 }

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 是任务的核心执行逻辑。
@@ -8,8 +13,8 @@ type Task interface {
 	// 返回的 error 表示任务是否执行成功。调度器会根据返回的 error 是否为 nil 来决定任务状态。
 	Execute() error
 
-	// ParseParams 解析参数
-	ParseParams() error
+	// ParseParams 解析及校验参数
+	ParseParams(logger *logs.Logger, claimedLog *models.TaskExecutionLog) error
 
 	// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。
 	// log: 任务执行的上下文。

internal/infra/utils/time.go

@@ -0,0 +1,22 @@
+package utils
+
+import (
+	"time"
+
+	"github.com/robfig/cron/v3"
+)
+
+// GetNextCronTime 根据传入的 Cron 表达式计算下一次执行的时间。
+// 它使用兼容大多数 Cron 实现的标准解析器。
+// 如果 Cron 表达式无效，它将返回一个错误。
+func GetNextCronTime(cronExpression string) (time.Time, error) {
+	// cron.ParseStandard() 返回一个支持标准5位或6位（带秒）格式的解析器。
+	schedule, err := cron.ParseStandard(cronExpression)
+	if err != nil {
+		return time.Time{}, err // 返回零值时间和错误
+	}
+
+	// 从当前时间计算下一次执行时间
+	nextTime := schedule.Next(time.Now())
+	return nextTime, nil
+}