重构名字
This commit is contained in:
316
internal/domain/plan/analysis_plan_task_manager.go
Normal file
316
internal/domain/plan/analysis_plan_task_manager.go
Normal file
@@ -0,0 +1,316 @@
|
||||
package plan
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"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"
|
||||
)
|
||||
|
||||
// AnalysisPlanTaskManager 负责管理分析计划的触发器任务。
|
||||
// 它确保数据库中可执行的计划在待执行队列中有对应的触发器，并移除无效的触发器。
|
||||
// 这是一个有状态的组件，包含一个互斥锁以确保并发安全。
|
||||
type AnalysisPlanTaskManager struct {
|
||||
planRepo repository.PlanRepository
|
||||
pendingTaskRepo repository.PendingTaskRepository
|
||||
executionLogRepo repository.ExecutionLogRepository
|
||||
logger *logs.Logger
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NewAnalysisPlanTaskManager 是 AnalysisPlanTaskManager 的构造函数。
|
||||
func NewAnalysisPlanTaskManager(
|
||||
planRepo repository.PlanRepository,
|
||||
pendingTaskRepo repository.PendingTaskRepository,
|
||||
executionLogRepo repository.ExecutionLogRepository,
|
||||
logger *logs.Logger,
|
||||
) *AnalysisPlanTaskManager {
|
||||
return &AnalysisPlanTaskManager{
|
||||
planRepo: planRepo,
|
||||
pendingTaskRepo: pendingTaskRepo,
|
||||
executionLogRepo: executionLogRepo,
|
||||
logger: logger,
|
||||
}
|
||||
}
|
||||
|
||||
// Refresh 同步数据库中的计划状态和待执行队列中的触发器任务。
|
||||
// 这是一个编排方法，将复杂的逻辑分解到多个内部方法中。
|
||||
func (m *AnalysisPlanTaskManager) Refresh() error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
m.logger.Info("开始同步计划任务管理器...")
|
||||
|
||||
// 1. 一次性获取所有需要的数据
|
||||
runnablePlans, invalidPlanIDs, pendingTasks, err := m.getRefreshData()
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取刷新数据失败: %w", err)
|
||||
}
|
||||
|
||||
// 2. 清理所有与失效计划相关的待执行任务
|
||||
if err := m.cleanupInvalidTasks(invalidPlanIDs, pendingTasks); err != nil {
|
||||
// 仅记录错误，清理失败不应阻止新任务的添加
|
||||
m.logger.Errorf("清理无效任务时出错: %v", err)
|
||||
}
|
||||
|
||||
// 3. 添加或更新触发器
|
||||
if err := m.addOrUpdateTriggers(runnablePlans, pendingTasks); err != nil {
|
||||
return fmt.Errorf("添加或更新触发器时出错: %w", err)
|
||||
}
|
||||
|
||||
m.logger.Info("计划任务管理器同步完成.")
|
||||
return nil
|
||||
}
|
||||
|
||||
// CreateOrUpdateTrigger 为给定的 planID 创建其关联的触发任务。
|
||||
// 如果触发器已存在，会根据计划类型更新其执行时间。
|
||||
func (m *AnalysisPlanTaskManager) CreateOrUpdateTrigger(planID uint) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
// 检查计划是否可执行
|
||||
plan, err := m.planRepo.GetBasicPlanByID(planID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("获取计划基本信息失败: %w", err)
|
||||
}
|
||||
if plan.Status != models.PlanStatusEnabled {
|
||||
return fmt.Errorf("计划 #%d 当前状态为 '%d'，无法创建或更新触发器", planID, plan.Status)
|
||||
}
|
||||
|
||||
// 查找现有触发器
|
||||
existingTrigger, err := m.pendingTaskRepo.FindPendingTriggerByPlanID(planID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("查找现有触发器失败: %w", err)
|
||||
}
|
||||
|
||||
// 如果触发器已存在，则根据计划类型更新其执行时间
|
||||
if existingTrigger != nil {
|
||||
var expectedExecuteAt time.Time
|
||||
if plan.ExecutionType == models.PlanExecutionTypeManual {
|
||||
// 手动计划，如果再次触发，则立即执行
|
||||
expectedExecuteAt = time.Now()
|
||||
} else { // 自动计划
|
||||
// 自动计划，根据 Cron 表达式计算下一次执行时间
|
||||
next, err := utils.GetNextCronTime(plan.CronExpression)
|
||||
if err != nil {
|
||||
m.logger.Errorf("为计划 #%d 解析Cron表达式失败，无法更新触发器: %v", plan.ID, err)
|
||||
return fmt.Errorf("解析 Cron 表达式失败: %w", err)
|
||||
}
|
||||
expectedExecuteAt = next
|
||||
}
|
||||
|
||||
// 如果计算出的执行时间与当前待执行任务的时间不一致，则更新
|
||||
if !existingTrigger.ExecuteAt.Equal(expectedExecuteAt) {
|
||||
m.logger.Infof("计划 #%d 的执行时间已变更，正在更新触发器 #%d 的执行时间从 %v 到 %v...", plan.ID, existingTrigger.ID, existingTrigger.ExecuteAt, expectedExecuteAt)
|
||||
if err := m.pendingTaskRepo.UpdatePendingTaskExecuteAt(existingTrigger.ID, expectedExecuteAt); err != nil {
|
||||
m.logger.Errorf("更新触发器 #%d 的执行时间失败: %v", existingTrigger.ID, err)
|
||||
return fmt.Errorf("更新触发器执行时间失败: %w", err)
|
||||
}
|
||||
} else {
|
||||
m.logger.Infof("计划 #%d 的触发器已存在且执行时间无需更新。", plan.ID)
|
||||
}
|
||||
return nil // 触发器已存在且已处理更新，直接返回
|
||||
}
|
||||
|
||||
// 如果触发器不存在，则创建新的触发器
|
||||
m.logger.Infof("为计划 #%d 创建新的触发器...", planID)
|
||||
return m.createTriggerTask(plan)
|
||||
}
|
||||
|
||||
// EnsureAnalysisTaskDefinition 确保计划的分析任务定义存在于 tasks 表中。
|
||||
// 如果不存在，则会自动创建。此方法不涉及待执行队列。
|
||||
func (m *AnalysisPlanTaskManager) EnsureAnalysisTaskDefinition(planID uint) error {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
|
||||
plan, err := m.planRepo.GetBasicPlanByID(planID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("确保分析任务定义失败：获取计划 #%d 基本信息时出错: %w", planID, err)
|
||||
}
|
||||
|
||||
analysisTask, err := m.planRepo.FindPlanAnalysisTaskByPlanID(plan.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("确保分析任务定义失败：查找计划 #%d 的分析任务时出错: %w", plan.ID, err)
|
||||
}
|
||||
|
||||
if analysisTask == nil {
|
||||
m.logger.Infof("未找到计划 #%d 关联的 'plan_analysis' 任务定义，将自动创建...", plan.ID)
|
||||
_, err := m.planRepo.CreatePlanAnalysisTask(plan) // CreatePlanAnalysisTask returns *models.Task, error
|
||||
if err != nil {
|
||||
return fmt.Errorf("自动创建 'plan_analysis' 任务定义失败: %w", err)
|
||||
}
|
||||
m.logger.Infof("已成功为计划 #%d 创建 'plan_analysis' 任务定义。", plan.ID)
|
||||
} else {
|
||||
m.logger.Infof("计划 #%d 的 'plan_analysis' 任务定义已存在。", plan.ID)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// --- 内部私有方法 ---
|
||||
|
||||
// getRefreshData 从数据库获取刷新所需的所有数据。
|
||||
func (m *AnalysisPlanTaskManager) getRefreshData() (runnablePlans []*models.Plan, invalidPlanIDs []uint, pendingTasks []models.PendingTask, err error) {
|
||||
runnablePlans, err = m.planRepo.FindRunnablePlans()
|
||||
if err != nil {
|
||||
m.logger.Errorf("获取可执行计划列表失败: %v", err)
|
||||
return
|
||||
}
|
||||
|
||||
invalidPlans, err := m.planRepo.FindInactivePlans()
|
||||
if err != nil {
|
||||
m.logger.Errorf("获取失效计划列表失败: %v", err)
|
||||
return
|
||||
}
|
||||
invalidPlanIDs = make([]uint, len(invalidPlans))
|
||||
for i, p := range invalidPlans {
|
||||
invalidPlanIDs[i] = p.ID
|
||||
}
|
||||
|
||||
pendingTasks, err = m.pendingTaskRepo.FindAllPendingTasks()
|
||||
if err != nil {
|
||||
m.logger.Errorf("获取所有待执行任务失败: %v", err)
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
// cleanupInvalidTasks 清理所有与失效计划相关的待执行任务。
|
||||
func (m *AnalysisPlanTaskManager) cleanupInvalidTasks(invalidPlanIDs []uint, allPendingTasks []models.PendingTask) error {
|
||||
if len(invalidPlanIDs) == 0 {
|
||||
return nil // 没有需要清理的计划
|
||||
}
|
||||
|
||||
invalidPlanIDSet := make(map[uint]struct{}, len(invalidPlanIDs))
|
||||
for _, id := range invalidPlanIDs {
|
||||
invalidPlanIDSet[id] = struct{}{}
|
||||
}
|
||||
|
||||
var tasksToDeleteIDs []uint
|
||||
var logsToCancelIDs []uint
|
||||
|
||||
for _, pt := range allPendingTasks {
|
||||
if pt.Task == nil { // 防御性编程，确保 Task 被预加载
|
||||
continue
|
||||
}
|
||||
if _, isInvalid := invalidPlanIDSet[pt.Task.PlanID]; isInvalid {
|
||||
tasksToDeleteIDs = append(tasksToDeleteIDs, pt.ID)
|
||||
logsToCancelIDs = append(logsToCancelIDs, pt.TaskExecutionLogID)
|
||||
}
|
||||
}
|
||||
|
||||
if len(tasksToDeleteIDs) == 0 {
|
||||
return nil // 没有找到需要清理的任务
|
||||
}
|
||||
|
||||
m.logger.Infof("准备从待执行队列中清理 %d 个与失效计划相关的任务...", len(tasksToDeleteIDs))
|
||||
|
||||
// 批量删除待执行任务
|
||||
if err := m.pendingTaskRepo.DeletePendingTasksByIDs(tasksToDeleteIDs); err != nil {
|
||||
return fmt.Errorf("批量删除待执行任务失败: %w", err)
|
||||
}
|
||||
|
||||
// 批量更新相关执行日志状态为“已取消”
|
||||
if err := m.executionLogRepo.UpdateTaskExecutionLogStatusByIDs(logsToCancelIDs, models.ExecutionStatusCancelled); err != nil {
|
||||
// 这是一个非关键性错误，只记录日志
|
||||
m.logger.Warnf("批量更新日志状态为 'Cancelled' 失败: %v", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// addOrUpdateTriggers 检查、更新或创建触发器。
|
||||
func (m *AnalysisPlanTaskManager) addOrUpdateTriggers(runnablePlans []*models.Plan, allPendingTasks []models.PendingTask) error {
|
||||
// 创建一个映射，存放所有已在队列中的计划触发器
|
||||
pendingTriggersMap := make(map[uint]models.PendingTask)
|
||||
for _, pt := range allPendingTasks {
|
||||
if pt.Task != nil && pt.Task.Type == models.TaskPlanAnalysis {
|
||||
pendingTriggersMap[pt.Task.PlanID] = pt
|
||||
}
|
||||
}
|
||||
|
||||
for _, plan := range runnablePlans {
|
||||
existingTrigger, exists := pendingTriggersMap[plan.ID]
|
||||
|
||||
if exists {
|
||||
// --- 新增逻辑：检查并更新现有触发器 ---
|
||||
// 只对自动计划检查时间更新
|
||||
if plan.ExecutionType == models.PlanExecutionTypeAutomatic {
|
||||
next, err := utils.GetNextCronTime(plan.CronExpression)
|
||||
if err != nil {
|
||||
m.logger.Errorf("为计划 #%d 解析Cron表达式失败，跳过更新: %v", plan.ID, err)
|
||||
continue
|
||||
}
|
||||
// 如果数据库中记录的执行时间与根据当前Cron表达式计算出的下一次时间不一致，则更新
|
||||
if !existingTrigger.ExecuteAt.Equal(next) {
|
||||
m.logger.Infof("计划 #%d 的执行时间已变更，正在更新触发器 #%d 的执行时间从 %v 到 %v...", plan.ID, existingTrigger.ID, existingTrigger.ExecuteAt, next)
|
||||
if err := m.pendingTaskRepo.UpdatePendingTaskExecuteAt(existingTrigger.ID, next); err != nil {
|
||||
m.logger.Errorf("更新触发器 #%d 的执行时间失败: %v", existingTrigger.ID, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// --- 原有逻辑：为缺失的计划创建新触发器 ---
|
||||
m.logger.Infof("发现应执行但队列中缺失的计划 #%d，正在为其创建触发器...", plan.ID)
|
||||
if err := m.createTriggerTask(plan); err != nil {
|
||||
m.logger.Errorf("为计划 #%d 创建触发器失败: %v", plan.ID, err)
|
||||
// 继续处理下一个，不因单点失败而中断
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// createTriggerTask 是创建触发器任务的内部核心逻辑。
|
||||
func (m *AnalysisPlanTaskManager) createTriggerTask(plan *models.Plan) error {
|
||||
analysisTask, err := m.planRepo.FindPlanAnalysisTaskByPlanID(plan.ID)
|
||||
if err != nil {
|
||||
return fmt.Errorf("查找计划分析任务失败: %w", err)
|
||||
}
|
||||
|
||||
// --- 如果触发器任务定义不存在，则自动创建 ---
|
||||
if analysisTask == nil {
|
||||
m.logger.Warnf("未找到计划 #%d 关联的 'plan_analysis' 任务定义，将自动创建...", plan.ID)
|
||||
newAnalysisTask, err := m.planRepo.CreatePlanAnalysisTask(plan)
|
||||
if err != nil {
|
||||
return fmt.Errorf("自动创建 'plan_analysis' 任务定义失败: %w", err)
|
||||
}
|
||||
analysisTask = newAnalysisTask
|
||||
m.logger.Infof("已成功为计划 #%d 创建 'plan_analysis' 任务定义 (ID: %d)", plan.ID, analysisTask.ID)
|
||||
}
|
||||
|
||||
var executeAt time.Time
|
||||
if plan.ExecutionType == models.PlanExecutionTypeManual {
|
||||
executeAt = time.Now()
|
||||
} else {
|
||||
next, err := utils.GetNextCronTime(plan.CronExpression)
|
||||
if err != nil {
|
||||
return fmt.Errorf("解析 Cron 表达式 '%s' 失败: %w", plan.CronExpression, err)
|
||||
}
|
||||
executeAt = next
|
||||
}
|
||||
|
||||
taskLog := &models.TaskExecutionLog{
|
||||
TaskID: analysisTask.ID,
|
||||
Status: models.ExecutionStatusWaiting,
|
||||
}
|
||||
if err := m.executionLogRepo.CreateTaskExecutionLog(taskLog); err != nil {
|
||||
return fmt.Errorf("创建任务执行日志失败: %w", err)
|
||||
}
|
||||
|
||||
pendingTask := &models.PendingTask{
|
||||
TaskID: analysisTask.ID,
|
||||
ExecuteAt: executeAt,
|
||||
TaskExecutionLogID: taskLog.ID,
|
||||
}
|
||||
if err := m.pendingTaskRepo.CreatePendingTask(pendingTask); err != nil {
|
||||
return fmt.Errorf("创建待执行任务失败: %w", err)
|
||||
}
|
||||
|
||||
m.logger.Infof("成功为计划 #%d 创建触发器 (任务ID: %d)，执行时间: %v", plan.ID, analysisTask.ID, executeAt)
|
||||
return nil
|
||||
}
|
||||
466
internal/domain/plan/plan_execution_manager.go
Normal file
466
internal/domain/plan/plan_execution_manager.go
Normal file
@@ -0,0 +1,466 @@
|
||||
package plan
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.huangwc.com/pig/pig-farm-controller/internal/domain/device"
|
||||
"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"
|
||||
)
|
||||
|
||||
// ProgressTracker 仅用于在内存中提供计划执行的并发锁
|
||||
type ProgressTracker struct {
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond // 用于实现阻塞锁
|
||||
runningPlans map[uint]bool // key: planExecutionLogID, value: true (用作内存锁)
|
||||
}
|
||||
|
||||
// NewProgressTracker 创建一个新的进度跟踪器
|
||||
func NewProgressTracker() *ProgressTracker {
|
||||
t := &ProgressTracker{
|
||||
runningPlans: make(map[uint]bool),
|
||||
}
|
||||
t.cond = sync.NewCond(&t.mu)
|
||||
return t
|
||||
}
|
||||
|
||||
// TryLock (非阻塞) 尝试锁定一个计划。如果计划未被锁定，则锁定并返回 true。
|
||||
func (t *ProgressTracker) TryLock(planLogID uint) bool {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
if t.runningPlans[planLogID] {
|
||||
return false // 已被锁定
|
||||
}
|
||||
t.runningPlans[planLogID] = true
|
||||
return true
|
||||
}
|
||||
|
||||
// Lock (阻塞) 获取一个计划的执行锁。如果锁已被占用，则会一直等待直到锁被释放。
|
||||
func (t *ProgressTracker) Lock(planLogID uint) {
|
||||
t.mu.Lock()
|
||||
// 当计划正在运行时，调用 t.cond.Wait() 会原子地解锁 mu 并挂起当前协程。
|
||||
// 当被唤醒时，它会重新锁定 mu 并再次检查循环条件。
|
||||
for t.runningPlans[planLogID] {
|
||||
t.cond.Wait()
|
||||
}
|
||||
// 获取到锁
|
||||
t.runningPlans[planLogID] = true
|
||||
t.mu.Unlock()
|
||||
}
|
||||
|
||||
// Unlock 解锁一个计划，并唤醒所有正在等待此锁的协程。
|
||||
func (t *ProgressTracker) Unlock(planLogID uint) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
delete(t.runningPlans, planLogID)
|
||||
// 唤醒所有在此条件上等待的协程
|
||||
t.cond.Broadcast()
|
||||
}
|
||||
|
||||
// GetRunningPlanIDs 获取当前所有正在执行的计划ID列表
|
||||
func (t *ProgressTracker) GetRunningPlanIDs() []uint {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
ids := make([]uint, 0, len(t.runningPlans))
|
||||
for id := range t.runningPlans {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids
|
||||
}
|
||||
|
||||
// PlanExecutionManager 是核心的、持久化的任务调度器
|
||||
type PlanExecutionManager struct {
|
||||
logger *logs.Logger
|
||||
pollingInterval time.Duration
|
||||
workers int
|
||||
pendingTaskRepo repository.PendingTaskRepository
|
||||
executionLogRepo repository.ExecutionLogRepository
|
||||
deviceRepo repository.DeviceRepository
|
||||
sensorDataRepo repository.SensorDataRepository
|
||||
planRepo repository.PlanRepository
|
||||
taskFactory TaskFactory
|
||||
analysisPlanTaskManager *AnalysisPlanTaskManager
|
||||
progressTracker *ProgressTracker
|
||||
deviceService device.Service
|
||||
|
||||
pool *ants.Pool // 使用 ants 协程池来管理并发
|
||||
wg sync.WaitGroup
|
||||
stopChan chan struct{} // 用于停止主循环的信号通道
|
||||
}
|
||||
|
||||
// NewPlanExecutionManager 创建一个新的调度器实例
|
||||
func NewPlanExecutionManager(
|
||||
pendingTaskRepo repository.PendingTaskRepository,
|
||||
executionLogRepo repository.ExecutionLogRepository,
|
||||
deviceRepo repository.DeviceRepository,
|
||||
sensorDataRepo repository.SensorDataRepository,
|
||||
planRepo repository.PlanRepository,
|
||||
analysisPlanTaskManager *AnalysisPlanTaskManager,
|
||||
taskFactory TaskFactory,
|
||||
logger *logs.Logger,
|
||||
deviceService device.Service,
|
||||
interval time.Duration,
|
||||
numWorkers int,
|
||||
) *PlanExecutionManager {
|
||||
return &PlanExecutionManager{
|
||||
pendingTaskRepo: pendingTaskRepo,
|
||||
executionLogRepo: executionLogRepo,
|
||||
deviceRepo: deviceRepo,
|
||||
sensorDataRepo: sensorDataRepo,
|
||||
planRepo: planRepo,
|
||||
analysisPlanTaskManager: analysisPlanTaskManager,
|
||||
taskFactory: taskFactory,
|
||||
logger: logger,
|
||||
deviceService: deviceService,
|
||||
pollingInterval: interval,
|
||||
workers: numWorkers,
|
||||
progressTracker: NewProgressTracker(),
|
||||
stopChan: make(chan struct{}), // 初始化停止信号通道
|
||||
}
|
||||
}
|
||||
|
||||
// Start 启动调度器，包括初始化协程池和启动主轮询循环
|
||||
func (s *PlanExecutionManager) Start() {
|
||||
s.logger.Warnf("任务调度器正在启动，工作协程数: %d...", s.workers)
|
||||
pool, err := ants.NewPool(s.workers, ants.WithPanicHandler(func(err interface{}) {
|
||||
s.logger.Errorf("[严重] 任务执行时发生 panic: %v", err)
|
||||
}))
|
||||
if err != nil {
|
||||
panic("初始化协程池失败: " + err.Error())
|
||||
}
|
||||
s.pool = pool
|
||||
|
||||
s.wg.Add(1)
|
||||
go s.run()
|
||||
s.logger.Warnf("任务调度器已成功启动")
|
||||
}
|
||||
|
||||
// Stop 优雅地停止调度器
|
||||
func (s *PlanExecutionManager) Stop() {
|
||||
s.logger.Warnf("正在停止任务调度器...")
|
||||
close(s.stopChan) // 1. 发出停止信号，停止主循环
|
||||
s.wg.Wait() // 2. 等待主循环完成
|
||||
s.pool.Release() // 3. 释放 ants 池 (等待所有已提交的任务执行完毕)
|
||||
s.logger.Warnf("任务调度器已安全停止")
|
||||
}
|
||||
|
||||
// run 是主轮询循环，负责从数据库认领任务并提交到协程池
|
||||
func (s *PlanExecutionManager) run() {
|
||||
defer s.wg.Done()
|
||||
ticker := time.NewTicker(s.pollingInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-s.stopChan:
|
||||
// 收到停止信号，退出循环
|
||||
return
|
||||
case <-ticker.C:
|
||||
// 定时触发任务认领和提交
|
||||
go s.claimAndSubmit()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// claimAndSubmit 实现了最终的“认领-锁定-执行 或 等待-放回”的健壮逻辑
|
||||
func (s *PlanExecutionManager) claimAndSubmit() {
|
||||
runningPlanIDs := s.progressTracker.GetRunningPlanIDs()
|
||||
|
||||
claimedLog, pendingTask, err := s.pendingTaskRepo.ClaimNextAvailableTask(runningPlanIDs)
|
||||
if err != nil {
|
||||
if !errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
s.logger.Errorf("认领任务时发生错误: %v", err)
|
||||
}
|
||||
// gorm.ErrRecordNotFound 说明没任务要执行
|
||||
return
|
||||
}
|
||||
|
||||
// 尝试获取内存执行锁
|
||||
if s.progressTracker.TryLock(claimedLog.PlanExecutionLogID) {
|
||||
// 成功获取锁，正常派发任务
|
||||
err = s.pool.Submit(func() {
|
||||
defer s.progressTracker.Unlock(claimedLog.PlanExecutionLogID)
|
||||
s.processTask(claimedLog)
|
||||
})
|
||||
if err != nil {
|
||||
s.logger.Errorf("向协程池提交任务失败: %v", err)
|
||||
// 提交失败，必须释放刚刚获取的锁
|
||||
s.progressTracker.Unlock(claimedLog.PlanExecutionLogID)
|
||||
// 同样需要将任务安全放回
|
||||
s.handleRequeue(claimedLog.PlanExecutionLogID, pendingTask)
|
||||
}
|
||||
} else {
|
||||
// 获取锁失败，说明有“兄弟”任务正在执行。执行“锁定并安全放回”逻辑。
|
||||
s.handleRequeue(claimedLog.PlanExecutionLogID, pendingTask)
|
||||
}
|
||||
}
|
||||
|
||||
// handleRequeue 同步地、安全地将一个无法立即执行的任务放回队列。
|
||||
func (s *PlanExecutionManager) handleRequeue(planExecutionLogID uint, taskToRequeue *models.PendingTask) {
|
||||
s.logger.Warnf("计划 %d 正在执行，任务 %d (TaskID: %d) 将等待并重新入队...", planExecutionLogID, taskToRequeue.ID, taskToRequeue.TaskID)
|
||||
|
||||
// 1. 阻塞式地等待，直到可以获取到该计划的锁。
|
||||
s.progressTracker.Lock(planExecutionLogID)
|
||||
defer s.progressTracker.Unlock(planExecutionLogID)
|
||||
|
||||
// 2. 在持有锁的情况下，将任务安全地放回队列。
|
||||
if err := s.pendingTaskRepo.RequeueTask(taskToRequeue); err != nil {
|
||||
s.logger.Errorf("[严重] 任务重新入队失败, 原始PendingTaskID: %d, 错误: %v", taskToRequeue.ID, err)
|
||||
return
|
||||
}
|
||||
|
||||
s.logger.Warnf("任务 (原始ID: %d) 已成功重新入队，并已释放计划 %d 的锁。", taskToRequeue.ID, planExecutionLogID)
|
||||
}
|
||||
|
||||
// processTask 处理单个任务的逻辑
|
||||
func (s *PlanExecutionManager) processTask(claimedLog *models.TaskExecutionLog) {
|
||||
s.logger.Warnf("开始处理任务, 日志ID: %d, 任务ID: %d, 任务名称: %s, 描述: %s",
|
||||
claimedLog.ID, claimedLog.TaskID, claimedLog.Task.Name, claimedLog.Task.Description)
|
||||
|
||||
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()
|
||||
|
||||
// 任务失败时，调用统一的终止服务
|
||||
s.handlePlanTermination(claimedLog.PlanExecutionLogID, "子任务执行失败: "+err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
// 如果是计划分析任务，它的职责是解析和分发任务，到此即完成，不参与后续的计划完成度检查。
|
||||
if claimedLog.Task.Type == models.TaskPlanAnalysis {
|
||||
s.logger.Warnf("完成计划分析任务, 日志ID: %d", claimedLog.ID)
|
||||
return
|
||||
}
|
||||
|
||||
// --- 以下是常规任务的完成逻辑 ---
|
||||
s.logger.Warnf("完成任务, 日志ID: %d", claimedLog.ID)
|
||||
|
||||
// 检查是否是最后一个任务
|
||||
incompleteCount, err := s.executionLogRepo.CountIncompleteTasksByPlanLogID(claimedLog.PlanExecutionLogID)
|
||||
if err != nil {
|
||||
s.logger.Errorf("检查计划 %d 的未完成任务数时出错: %v", claimedLog.PlanExecutionLogID, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 如果此计划执行中，未完成的任务只剩下当前这一个（因为当前任务的状态此时在数据库中仍为 'started'），
|
||||
// 则认为整个计划已完成。
|
||||
if incompleteCount == 1 {
|
||||
s.handlePlanCompletion(claimedLog.PlanExecutionLogID)
|
||||
}
|
||||
}
|
||||
|
||||
// runTask 用于执行具体任务
|
||||
func (s *PlanExecutionManager) 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.Production(claimedLog)
|
||||
|
||||
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 *PlanExecutionManager) analysisPlan(claimedLog *models.TaskExecutionLog) error {
|
||||
// 创建Plan执行记录
|
||||
// 从任务的 Parameters 中解析出真实的 PlanID
|
||||
var params struct {
|
||||
PlanID uint `json:"plan_id"`
|
||||
}
|
||||
if err := claimedLog.Task.ParseParameters(¶ms); err != nil {
|
||||
s.logger.Errorf("解析任务参数中的计划ID失败，日志ID: %d, 错误: %v", claimedLog.ID, err)
|
||||
return err
|
||||
}
|
||||
realPlanID := params.PlanID
|
||||
|
||||
planLog := &models.PlanExecutionLog{
|
||||
PlanID: realPlanID, // 使用从参数中解析出的真实 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(realPlanID)
|
||||
if err != nil {
|
||||
s.logger.Errorf("[严重] 解析计划失败, 日志ID: %d, 错误: %v", claimedLog.ID, err)
|
||||
return err
|
||||
}
|
||||
|
||||
// 写入执行历史
|
||||
taskLogs := make([]*models.TaskExecutionLog, len(tasks))
|
||||
for i, task := range tasks {
|
||||
taskLogs[i] = &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[i] = &models.PendingTask{
|
||||
TaskID: task.ID,
|
||||
TaskExecutionLogID: taskLogs[i].ID, // 使用正确的 TaskExecutionLogID
|
||||
|
||||
// 待执行队列是通过任务触发时间排序的, 且只要在调度器获取的时间点之前的都可以被触发
|
||||
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
|
||||
}
|
||||
|
||||
// --- 处理空计划的边缘情况 ---
|
||||
// 如果一个计划被解析后，发现其任务列表为空，
|
||||
// 那么它实际上已经“执行”完毕了，我们需要在这里手动为它创建下一次的触发器。
|
||||
if len(tasks) == 0 {
|
||||
s.handlePlanCompletion(planLog.ID)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// updateTaskExecutionLogStatus 修改任务历史中的执行状态
|
||||
func (s *PlanExecutionManager) 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
|
||||
}
|
||||
|
||||
// handlePlanTermination 集中处理计划的终止逻辑（失败或取消）
|
||||
func (s *PlanExecutionManager) handlePlanTermination(planLogID uint, reason string) {
|
||||
// 1. 从待执行队列中删除所有相关的子任务
|
||||
if err := s.pendingTaskRepo.DeletePendingTasksByPlanLogID(planLogID); err != nil {
|
||||
s.logger.Errorf("从待执行队列中删除计划 %d 的后续任务时出错: %v", planLogID, err)
|
||||
}
|
||||
|
||||
// 2. 将父计划的执行日志标记为失败
|
||||
if err := s.executionLogRepo.FailPlanExecution(planLogID, reason); err != nil {
|
||||
s.logger.Errorf("标记计划执行日志 %d 为失败时出错: %v", planLogID, err)
|
||||
}
|
||||
|
||||
// 3. 将所有未完成的子任务日志标记为已取消
|
||||
if err := s.executionLogRepo.CancelIncompleteTasksByPlanLogID(planLogID, "父计划失败或被取消"); err != nil {
|
||||
s.logger.Errorf("取消计划 %d 的后续任务日志时出错: %v", planLogID, err)
|
||||
}
|
||||
|
||||
// 4. 获取计划执行日志以获取顶层 PlanID
|
||||
planLog, err := s.executionLogRepo.FindPlanExecutionLogByID(planLogID)
|
||||
if err != nil {
|
||||
s.logger.Errorf("无法找到计划执行日志 %d 以更新父计划状态: %v", planLogID, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 5. 获取顶层计划的详细信息，以检查其类型
|
||||
topLevelPlan, err := s.planRepo.GetBasicPlanByID(planLog.PlanID)
|
||||
if err != nil {
|
||||
s.logger.Errorf("获取顶层计划 %d 的基本信息失败: %v", planLog.PlanID, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 6. 如果是系统任务，则不修改计划状态
|
||||
if topLevelPlan.PlanType == models.PlanTypeSystem {
|
||||
s.logger.Warnf("系统任务 %d (日志ID: %d) 执行失败，但根据策略不修改其计划状态。", topLevelPlan.ID, planLogID)
|
||||
return
|
||||
}
|
||||
|
||||
// 7. 将计划本身的状态更新为失败 (仅对非系统任务执行)
|
||||
if err := s.planRepo.UpdatePlanStatus(planLog.PlanID, models.PlanStatusFailed); err != nil {
|
||||
s.logger.Errorf("更新计划 %d 状态为 '失败' 时出错: %v", planLog.PlanID, err)
|
||||
}
|
||||
}
|
||||
|
||||
// handlePlanCompletion 集中处理计划成功完成后的所有逻辑
|
||||
func (s *PlanExecutionManager) handlePlanCompletion(planLogID uint) {
|
||||
s.logger.Infof("计划执行 %d 的所有任务已完成，开始处理计划完成逻辑...", planLogID)
|
||||
|
||||
// 1. 通过 PlanExecutionLog 反查正确的顶层 PlanID
|
||||
planExecutionLog, err := s.executionLogRepo.FindPlanExecutionLogByID(planLogID)
|
||||
if err != nil {
|
||||
s.logger.Errorf("获取计划执行日志 %d 失败: %v", planLogID, err)
|
||||
return
|
||||
}
|
||||
topLevelPlanID := planExecutionLog.PlanID // 这才是正确的顶层计划ID
|
||||
|
||||
// 2. 获取计划的最新数据，这里我们只需要基本信息来判断执行类型和次数
|
||||
plan, err := s.planRepo.GetBasicPlanByID(topLevelPlanID)
|
||||
if err != nil {
|
||||
s.logger.Errorf("获取计划 %d 的基本信息失败: %v", topLevelPlanID, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 3. 在内存中计算新的计数值和状态
|
||||
newExecuteCount := plan.ExecuteCount + 1
|
||||
newStatus := plan.Status // 默认为当前状态
|
||||
|
||||
// 如果是自动计划且达到执行次数上限，或计划是手动类型，则更新计划状态为已停止
|
||||
if (plan.ExecutionType == models.PlanExecutionTypeAutomatic && plan.ExecuteNum > 0 && newExecuteCount >= plan.ExecuteNum) || plan.ExecutionType == models.PlanExecutionTypeManual {
|
||||
newStatus = models.PlanStatusStopped
|
||||
s.logger.Infof("计划 %d 已完成执行，状态更新为 '执行完毕'。", topLevelPlanID)
|
||||
}
|
||||
|
||||
// 4. 使用专门的方法来原子性地更新计数值和状态
|
||||
if err := s.planRepo.UpdatePlanStateAfterExecution(topLevelPlanID, newExecuteCount, newStatus); err != nil {
|
||||
s.logger.Errorf("更新计划 %d 的执行后状态失败: %v", topLevelPlanID, err)
|
||||
return
|
||||
}
|
||||
|
||||
// 5. 更新计划执行日志状态为完成
|
||||
if err := s.executionLogRepo.UpdatePlanExecutionLogStatus(planLogID, models.ExecutionStatusCompleted); err != nil {
|
||||
s.logger.Errorf("更新计划执行日志 %d 状态为 '完成' 失败: %v", planLogID, err)
|
||||
}
|
||||
|
||||
// 6. 调用共享的 Manager 来处理触发器更新逻辑
|
||||
// 只有当计划在本次执行后仍然是 Enabled 状态时，才需要创建下一次的触发器。
|
||||
if newStatus == models.PlanStatusEnabled {
|
||||
if err := s.analysisPlanTaskManager.CreateOrUpdateTrigger(topLevelPlanID); err != nil {
|
||||
s.logger.Errorf("为计划 %d 创建/更新触发器失败: %v", topLevelPlanID, err)
|
||||
}
|
||||
} else {
|
||||
s.logger.Infof("计划 %d 状态为 '%d'，无需创建下一次触发器。", topLevelPlanID, newStatus)
|
||||
}
|
||||
}
|
||||
23
internal/domain/plan/task.go
Normal file
23
internal/domain/plan/task.go
Normal file
@@ -0,0 +1,23 @@
|
||||
package plan
|
||||
|
||||
import "git.huangwc.com/pig/pig-farm-controller/internal/infra/models"
|
||||
|
||||
// Task 定义了所有可被调度器执行的任务必须实现的接口。
|
||||
type Task interface {
|
||||
// Execute 是任务的核心执行逻辑。
|
||||
// ctx: 用于控制任务的超时或取消。
|
||||
// log: 包含了当前任务执行的完整上下文信息，包括从数据库中加载的任务参数等。
|
||||
// 返回的 error 表示任务是否执行成功。调度器会根据返回的 error 是否为 nil 来决定任务状态。
|
||||
Execute() error
|
||||
|
||||
// OnFailure 定义了当 Execute 方法返回错误时，需要执行的回滚或清理逻辑。
|
||||
// log: 任务执行的上下文。
|
||||
// executeErr: 从 Execute 方法返回的原始错误。
|
||||
OnFailure(executeErr error)
|
||||
}
|
||||
|
||||
// TaskFactory 是一个工厂接口，用于根据任务执行日志创建任务实例。
|
||||
type TaskFactory interface {
|
||||
// Production 根据指定的任务执行日志创建一个任务实例。
|
||||
Production(claimedLog *models.TaskExecutionLog) Task
|
||||
}
|
||||
Reference in New Issue
Block a user