task/task.go

// 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.Name()]; exist {
			return fmt.Errorf("%s 任务重复注册! ", taskInstance.Name())
		}
		d.taskTable[taskInstance.Name()] = 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
	)

	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.TaskName]; !exist {
		result.Err = fmt.Errorf("%v 任务未注册", cfg.TaskName)
	}
	d.lock.RUnlock()
	result.TaskRunID = taskInstance.GetRunID()
	result.TaskDescription = taskInstance.Description()
	if cfg.Async {
		// 异步运行
		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)
		}()
	} else {
		result.Data, result.Err = taskInstance.Execute(ctx, cfg)
		_ = taskInstance.Callback(result)
	}
	return result
}