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 +5 -151
4 changed files with 5 additions and 151 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit d6b415a839 - Show all commits

2
abstract.go
View File

@ -19,8 +19,6 @@ type ITask interface {
GetFlag() string
// Description 任务描述
Description() string
// GetRunID 获取任务ID
GetRunID() string
// Callback 任务执行成功的回调
Callback(result *Result) error
// Execute 执行任务

5
core.go
View File

@ -11,6 +11,7 @@ import (
"container/list"
"fmt"
"git.zhangdeman.cn/zhangdeman/easymap"
"git.zhangdeman.cn/zhangdeman/wrapper"
"time"
)
@ -172,6 +173,10 @@ func (tw *TimeWheel) checkAndRunTask() {
//
// Date : 13:43 2023/8/4
func (tw *TimeWheel) AddTask(task *Task, byInterval bool) {
if nil != task {
// 生成 key
task.Key = wrapper.StringFromRandom(128, "").Md5().Value
}
var pos, circle int
if byInterval {
pos, circle = tw.getPosAndCircleByInterval(task.Interval)

10
core_test.go
View File

@ -63,15 +63,5 @@ func (t testTask) Callback(result *Result) error {
func (t testTask) Execute(ctx context.Context, cfg *Config) (map[string]interface{}, error) {
fmt.Println(wrapper.OwnTimeFromNow().ToString())
tw.AddTask(&Task{
Key: wrapper.StringFromRandom(32, "").Value(),
Interval: 5 * time.Second,
CreatedTime: time.Now(),
Position: 0,
Circle: 100,
Job: &testTask{},
Times: 10,
MaxExecuteTime: 100,
}, false)
return nil, nil
}

139
task.go
View File

@ -1,139 +0,0 @@
// Package task ...
//
// Description : task ...
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 2022-06-23 15:13
package task
import (
"context"
"fmt"
"runtime"
"sync"
"time"
)
var (
// Dispatch 调度实例
Dispatch *dispatch
)
func init() {
Dispatch = &dispatch{
lock: &sync.RWMutex{},
taskTable: make(map[string]ITask),
}
}
// dispatch 任务调度
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:14 2022/6/23
type dispatch struct {
// 锁
lock *sync.RWMutex
// 任务表
taskTable map[string]ITask
}
// Register 注册任务
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:16 2022/6/23
func (d *dispatch) Register(taskInstanceList ...ITask) error {
d.lock.Lock()
defer d.lock.Unlock()
for _, taskInstance := range taskInstanceList {
if nil == taskInstance {
continue
}
if _, exist := d.taskTable[taskInstance.GetFlag()]; exist {
return fmt.Errorf("%s 任务重复注册! ", taskInstance.GetFlag())
}
d.taskTable[taskInstance.GetFlag()] = taskInstance
}
return nil
}
// Remove 移除任务
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:16 2022/6/23
func (d *dispatch) Remove(taskNameList ...string) error {
d.lock.Lock()
defer d.lock.Unlock()
for _, taskName := range taskNameList {
delete(d.taskTable, taskName)
}
return nil
}
// Run 执行任务
//
// Author : go_developer@163.com<白茶清欢>
//
// Date : 15:18 2022/6/23
func (d *dispatch) Run(ctx context.Context, cfg *Config) *Result {
var (
taskInstance ITask
exist bool
cancelFunc context.CancelFunc
)
if nil == ctx {
if cfg.Timeout > 0 {
ctx, cancelFunc = context.WithCancel(context.Background())
} else {
ctx = context.Background()
}
} else {
ctx, cancelFunc = context.WithCancel(ctx)
}
if nil != cancelFunc {
// 带了超时时间
cancelFunc()
}
result := &Result{
StartTime: time.Now().UnixNano(),
FinishTime: 0,
Used: 0,
TaskRunID: "",
TaskDescription: "",
TaskConfig: cfg,
Data: nil,
Err: nil,
}
defer func() {
result.FinishTime = time.Now().UnixNano()
result.Used = result.FinishTime - result.StartTime
}()
d.lock.RLock()
if taskInstance, exist = d.taskTable[cfg.TaskFlag]; !exist {
result.Err = fmt.Errorf("%v 任务未注册", cfg.TaskFlag)
return result
}
d.lock.RUnlock()
result.TaskRunID = taskInstance.GetRunID()
result.TaskDescription = taskInstance.Description()
// 异步运行
go func() {
if e := recover(); nil != e {
switch e.(type) {
case runtime.Error: // 运行时错误
result.Err = fmt.Errorf("出现运行时Panic : %v", e)
default: // 非运行时错误
result.Err = fmt.Errorf("出现其他场景Panic : %v", e)
}
}
result.Data, result.Err = taskInstance.Execute(ctx, cfg)
_ = taskInstance.Callback(result)
}()
return result
}