diff --git a/src/framework/cron/cron.go b/src/framework/cron/cron.go index ac14b05b..e835d375 100644 --- a/src/framework/cron/cron.go +++ b/src/framework/cron/cron.go @@ -63,7 +63,9 @@ type Queue struct { // QueueProcessor 队列处理函数接口 type QueueProcessor interface { // Execute 实际执行函数 - Execute(data any) any + // any 返回有效值最终序列化为字符串,记录为成功 + // error 存在错误,记录为失败 + Execute(data any) (any, error) } // RunJob 运行任务,data是传入的数据 @@ -196,7 +198,12 @@ func (s QueueJob) Run() { // 获取队列处理器接口实现 processor := *job.queueProcessor - result := processor.Execute(job.Data) - job.Status = Completed - newLog.Completed(result, "completed", job) + result, err := processor.Execute(job.Data) + if err != nil { + job.Status = Failed + newLog.Error(err, "failed", job) + } else { + job.Status = Completed + newLog.Completed(result, "completed", job) + } } diff --git a/src/framework/cron/cron_test.go b/src/framework/cron/cron_test.go index cb8f7776..207cfe16 100644 --- a/src/framework/cron/cron_test.go +++ b/src/framework/cron/cron_test.go @@ -19,10 +19,10 @@ var NewSimple = &Simple{} type Simple struct{} -func (s *Simple) Execute(data any) any { +func (s *Simple) Execute(data any) (any, error) { logger.Infof("执行=> %+v ", data) // 实现任务处理逻辑 - return data + return data, nil } func TestSimple(t *testing.T) { @@ -68,7 +68,7 @@ type FooProcessor struct { count int } -func (s *FooProcessor) Execute(data any) any { +func (s *FooProcessor) Execute(data any) (any, error) { logger.Infof("执行 %d %d => %+v ", s.count, s.progress, data) s.count++ @@ -85,7 +85,7 @@ func (s *FooProcessor) Execute(data any) any { // 改变任务进度 s.progress = i } - return data + return data, nil } func TestFoo(t *testing.T) { @@ -119,7 +119,7 @@ type BarProcessor struct { count int } -func (s *BarProcessor) Execute(data any) any { +func (s *BarProcessor) Execute(data any) (any, error) { logger.Infof("执行 %d %d => %+v ", s.count, s.progress, data) s.count++ @@ -145,7 +145,7 @@ func (s *BarProcessor) Execute(data any) any { s.progress = i } - return data + return data, nil } func TestBar(t *testing.T) { diff --git a/src/modules/monitor/processor/bar/bar.go b/src/modules/monitor/processor/bar/bar.go index ad2ace16..1d968923 100644 --- a/src/modules/monitor/processor/bar/bar.go +++ b/src/modules/monitor/processor/bar/bar.go @@ -20,7 +20,7 @@ type BarProcessor struct { count int } -func (s *BarProcessor) Execute(data any) any { +func (s *BarProcessor) Execute(data any) (any, error) { logger.Infof("执行 %d 次,上次进度: %d ", s.count, s.progress) s.count++ @@ -51,10 +51,11 @@ func (s *BarProcessor) Execute(data any) any { } // 返回结果,用于记录执行结果 - return map[string]any{ + result := map[string]any{ "repeat": options.Repeat, "jobName": sysJob.JobName, "invokeTarget": sysJob.InvokeTarget, "targetParams": sysJob.TargetParams, } + return result, nil } diff --git a/src/modules/monitor/processor/foo/foo.go b/src/modules/monitor/processor/foo/foo.go index ec6d6750..21dd527f 100644 --- a/src/modules/monitor/processor/foo/foo.go +++ b/src/modules/monitor/processor/foo/foo.go @@ -20,7 +20,7 @@ type FooProcessor struct { count int } -func (s *FooProcessor) Execute(data any) any { +func (s *FooProcessor) Execute(data any) (any, error) { logger.Infof("执行 %d 次,上次进度: %d ", s.count, s.progress) s.count++ @@ -43,10 +43,11 @@ func (s *FooProcessor) Execute(data any) any { } // 返回结果,用于记录执行结果 - return map[string]any{ + result := map[string]any{ "repeat": options.Repeat, "jobName": sysJob.JobName, "invokeTarget": sysJob.InvokeTarget, "targetParams": sysJob.TargetParams, } + return result, nil } diff --git a/src/modules/monitor/processor/simple/simple.go b/src/modules/monitor/processor/simple/simple.go index d0339376..1f997bfc 100644 --- a/src/modules/monitor/processor/simple/simple.go +++ b/src/modules/monitor/processor/simple/simple.go @@ -10,17 +10,18 @@ var NewProcessor = &simpleProcessor{} // simple 队列任务处理 type simpleProcessor struct{} -func (s *simpleProcessor) Execute(data any) any { +func (s *simpleProcessor) Execute(data any) (any, error) { options := data.(cron.JobData) sysJob := options.SysJob logger.Infof("重复 %v 任务ID %s", options.Repeat, sysJob.JobID) // 返回结果,用于记录执行结果 - return map[string]any{ + result := map[string]any{ "repeat": options.Repeat, "jobName": sysJob.JobName, "invokeTarget": sysJob.InvokeTarget, "targetParams": sysJob.TargetParams, } + return result, nil }