pig/pig-farm-controller
1
0
Fork 0
Code Issues 11 Pull Requests Actions Packages Projects Releases Wiki Activity

实现task工厂

Browse Source
This commit is contained in:
huang
2025-09-17 16:55:56 +08:00
parent 8b8c539e06
commit dde277c14d
7 changed files with 89 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
internal/app/service/task/delay_task.go
View File

@@ -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)

9
internal/app/service/task/plan_analysis_task.go
View File

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

17
internal/app/service/task/task.go Normal file
View File

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

1
internal/infra/models/plan.go
View File

@@ -29,6 +29,7 @@ const (
type TaskType string
const (
TaskPlanAnalysis TaskType = "plan_analysis" // 解析Plan的Task列表并添加到待执行队列的特殊任务
TaskTypeWaiting TaskType = "waiting" // 等待任务
)

29
internal/infra/task/plan_analysis_task.go Normal file
View File

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

51
internal/infra/task/scheduler.go
View File

@@ -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)
}

7
internal/infra/task/task.go
View File

@@ -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: 任务执行的上下文。