zhangdeman/task
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

第一版时间轮任务调度实现 #1

Merged
zhangdeman merged 15 commits from feature/timewheel into master 2023-09-04 20:13:17 +08:00
Conversation 0 Commits 15 Files Changed 8 +21 -24
3 changed files with 21 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a2ba1440d4 - Show all commits

6
abstract.go
View File

@ -15,8 +15,8 @@ import "context"
// //
// Date : 14:22 2022/6/23 // Date : 14:22 2022/6/23
type ITask interface { type ITask interface {
// Name 任务名称标识, 全局唯一 // GetFlag 任务名称标识, 全局唯一
Name() string GetFlag() string
// Description 任务描述 // Description 任务描述
Description() string Description() string
// GetRunID 获取任务ID // GetRunID 获取任务ID
@ -24,5 +24,5 @@ type ITask interface {
// Callback 任务执行成功的回调 // Callback 任务执行成功的回调
Callback(result *Result) error Callback(result *Result) error
// Execute 执行任务 // Execute 执行任务
Execute(ctx context.Context, cfg *Config) *Result Execute(ctx context.Context, cfg *Config) (map[string]interface{}, error)
} }

2
define.go
View File

@ -19,6 +19,8 @@ type Config struct {
ForbiddenCallback bool ForbiddenCallback bool
// Param 任务执行参数 // Param 任务执行参数
Param map[string]interface{} Param map[string]interface{}
// TaskFlag 任务标识
TaskFlag string
} }
// Result 执行结果 // Result 执行结果

17
task.go
View File

@ -51,10 +51,10 @@ func (d *dispatch) Register(taskInstanceList ...ITask) error {
if nil == taskInstance { if nil == taskInstance {
continue continue
} }
if _, exist := d.taskTable[taskInstance.Name()]; exist { if _, exist := d.taskTable[taskInstance.GetFlag()]; exist {
return fmt.Errorf("%s 任务重复注册! ", taskInstance.Name()) return fmt.Errorf("%s 任务重复注册! ", taskInstance.GetFlag())
} }
d.taskTable[taskInstance.Name()] = taskInstance d.taskTable[taskInstance.GetFlag()] = taskInstance
} }
return nil return nil
} }
@ -109,20 +109,19 @@ func (d *dispatch) Run(ctx context.Context, cfg *Config) *Result {
TaskConfig: cfg, TaskConfig: cfg,
Data: nil, Data: nil,
Err: nil, Err: nil,
Async: cfg.Async,
} }
defer func() { defer func() {
result.FinishTime = time.Now().UnixNano() result.FinishTime = time.Now().UnixNano()
result.Used = result.FinishTime - result.StartTime result.Used = result.FinishTime - result.StartTime
}() }()
d.lock.RLock() d.lock.RLock()
if taskInstance, exist = d.taskTable[cfg.TaskName]; !exist { if taskInstance, exist = d.taskTable[cfg.TaskFlag]; !exist {
result.Err = fmt.Errorf("%v 任务未注册", cfg.TaskName) result.Err = fmt.Errorf("%v 任务未注册", cfg.TaskFlag)
return result
} }
d.lock.RUnlock() d.lock.RUnlock()
result.TaskRunID = taskInstance.GetRunID() result.TaskRunID = taskInstance.GetRunID()
result.TaskDescription = taskInstance.Description() result.TaskDescription = taskInstance.Description()
if cfg.Async {
// 异步运行 // 异步运行
go func() { go func() {
if e := recover(); nil != e { if e := recover(); nil != e {
@ -136,9 +135,5 @@ func (d *dispatch) Run(ctx context.Context, cfg *Config) *Result {
result.Data, result.Err = taskInstance.Execute(ctx, cfg) result.Data, result.Err = taskInstance.Execute(ctx, cfg)
_ = taskInstance.Callback(result) _ = taskInstance.Callback(result)
}() }()
} else {
result.Data, result.Err = taskInstance.Execute(ctx, cfg)
_ = taskInstance.Callback(result)
}
return result return result
} }