init
This commit is contained in:
zyj
206
task/scheduler.go
Normal file
206
task/scheduler.go
Normal file
@@ -0,0 +1,206 @@
|
||||
package task
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"sort"
|
||||
"sync"
|
||||
|
||||
paramsTypes "go-account-register/types"
|
||||
|
||||
"github.com/fatih/structs"
|
||||
"github.com/google/uuid"
|
||||
)
|
||||
|
||||
type Scheduler struct {
|
||||
TaskQueue map[string]*paramsTypes.Task // 待处理的任务
|
||||
mu sync.Mutex
|
||||
ActiveProcesses map[string]*paramsTypes.Task // 正在运行的任务
|
||||
PausedTasks map[string]*paramsTypes.Task // 暂停的任务
|
||||
}
|
||||
|
||||
var (
|
||||
StopStatus string = "stop" // 暂停状态
|
||||
PendingStatus string = "pending" // 运行状态
|
||||
)
|
||||
|
||||
var MaxTaskNum int = 10
|
||||
|
||||
// 全局单例实例
|
||||
var globalScheduler *Scheduler
|
||||
var once sync.Once
|
||||
|
||||
// 获取全局调度器实例
|
||||
func GetScheduler() *Scheduler {
|
||||
once.Do(func() {
|
||||
globalScheduler = &Scheduler{
|
||||
TaskQueue: make(map[string]*paramsTypes.Task),
|
||||
ActiveProcesses: make(map[string]*paramsTypes.Task),
|
||||
PausedTasks: make(map[string]*paramsTypes.Task),
|
||||
}
|
||||
})
|
||||
return globalScheduler
|
||||
}
|
||||
|
||||
// 添加任务
|
||||
func (s *Scheduler) AddTask(params *paramsTypes.TaskParams) string {
|
||||
s.mu.Lock()
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
var uuidV4 string
|
||||
is := structs.New(params)
|
||||
if is.Field("ID").IsZero() {
|
||||
uuidV4 = uuid.NewString()
|
||||
} else {
|
||||
uuidV4 = params.ID
|
||||
}
|
||||
task := ¶msTypes.Task{
|
||||
ID: uuidV4,
|
||||
Name: params.Name,
|
||||
Status: params.Status,
|
||||
Cancel: cancel,
|
||||
Ctx: ctx,
|
||||
Message: make(chan string, 1),
|
||||
Type: params.Type,
|
||||
Data: ¶msTypes.TaskData{
|
||||
AccountClassify: params.AccountClassify,
|
||||
DataClassify: params.DataClassify,
|
||||
Keyword: params.Keyword,
|
||||
TwitterId: params.TwitterId,
|
||||
Compensate: params.Compensate,
|
||||
}}
|
||||
switch params.Status {
|
||||
case StopStatus:
|
||||
s.PausedTasks[uuidV4] = task
|
||||
case PendingStatus:
|
||||
s.TaskQueue[uuidV4] = task
|
||||
default:
|
||||
s.TaskQueue[uuidV4] = task
|
||||
}
|
||||
log.Println("创建任务：" + uuidV4)
|
||||
log.Println("状态：" + params.Status)
|
||||
s.mu.Unlock()
|
||||
s.processQueue()
|
||||
return uuidV4
|
||||
}
|
||||
|
||||
// 运行任务
|
||||
func (s *Scheduler) RunTask(id string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
// 查询暂停的任务列表中是否包含
|
||||
v, ok := s.PausedTasks[id]
|
||||
if ok {
|
||||
s.TaskQueue[id] = v
|
||||
delete(s.PausedTasks, id)
|
||||
}
|
||||
s.processQueue()
|
||||
}
|
||||
|
||||
// 暂停任务
|
||||
func (s *Scheduler) StopTask(id string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
// 查询任务列表中是否包含
|
||||
v, ok := s.TaskQueue[id]
|
||||
if ok {
|
||||
s.PausedTasks[id] = v
|
||||
v.Status = StopStatus
|
||||
v.Cancel() // 停止上下文
|
||||
// v.Message <- "Error"
|
||||
delete(s.TaskQueue, id)
|
||||
}
|
||||
// 查询运行中的任务列表中是否包含
|
||||
value, ok1 := s.ActiveProcesses[id]
|
||||
if ok1 {
|
||||
s.PausedTasks[id] = value
|
||||
value.Status = StopStatus
|
||||
log.Println("向通道中发送消息")
|
||||
value.Cancel() // 停止上下文
|
||||
// value.Message <- "Error"
|
||||
delete(s.ActiveProcesses, id)
|
||||
}
|
||||
}
|
||||
|
||||
// 删除任务
|
||||
func (s *Scheduler) RemoveTask(id string) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
// 查询任务列表中是否包含
|
||||
_, ok := s.TaskQueue[id]
|
||||
if ok {
|
||||
delete(s.TaskQueue, id)
|
||||
}
|
||||
|
||||
// 查询运行中的任务列表中是否包含
|
||||
_, ok1 := s.ActiveProcesses[id]
|
||||
if ok1 {
|
||||
delete(s.ActiveProcesses, id)
|
||||
}
|
||||
|
||||
// 查询暂停中的任务列表中是否包含
|
||||
_, ok2 := s.PausedTasks[id]
|
||||
if ok2 {
|
||||
delete(s.PausedTasks, id)
|
||||
}
|
||||
}
|
||||
|
||||
// 处理任务
|
||||
func (s *Scheduler) processQueue() {
|
||||
|
||||
s.mu.Lock()
|
||||
log.Println("开始处理任务")
|
||||
defer s.mu.Unlock()
|
||||
for len(s.ActiveProcesses) < MaxTaskNum && len(s.TaskQueue) > 0 {
|
||||
keys := make([]string, 0, len(s.TaskQueue))
|
||||
for k := range s.TaskQueue {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
|
||||
// 排序键
|
||||
sort.Strings(keys)
|
||||
// 获取第一个键值对
|
||||
firstKey := keys[0]
|
||||
go s.executeTask(s.TaskQueue[firstKey])
|
||||
s.ActiveProcesses[firstKey] = s.TaskQueue[firstKey]
|
||||
delete(s.TaskQueue, firstKey)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *Scheduler) executeTask(task *paramsTypes.Task) {
|
||||
// 根据类型判断任务
|
||||
// switch task.Type {
|
||||
// case 6:
|
||||
// var taskNurturingClashPlugin plugin.TaskNurturingClashPlugin
|
||||
// go taskNurturingClashPlugin.Run(task)
|
||||
// case 1:
|
||||
// var taskChangeAccoutnClashPlugin plugin.TaskChangeAccoutnClashPlugin
|
||||
// go taskChangeAccoutnClashPlugin.Run(task)
|
||||
// case 3:
|
||||
// var taskFollowClashPlugin plugin.TaskFollowClashPlugin
|
||||
// go taskFollowClashPlugin.Run(task)
|
||||
// case 7:
|
||||
// var taskSendGifPlugin plugin.TaskSendGifPlugin
|
||||
// go taskSendGifPlugin.Run(task)
|
||||
// }
|
||||
|
||||
// 创建一个阻塞获取当前任务
|
||||
select {
|
||||
case message := <-task.Message:
|
||||
if message == "Success" {
|
||||
log.Println("任务成功")
|
||||
// 运行结束
|
||||
s.PausedTasks[task.ID] = task
|
||||
delete(s.ActiveProcesses, task.ID)
|
||||
}
|
||||
if message == "Error" {
|
||||
log.Println("任务失败")
|
||||
// 运行结束
|
||||
s.PausedTasks[task.ID] = task
|
||||
delete(s.ActiveProcesses, task.ID)
|
||||
}
|
||||
case <-task.Ctx.Done():
|
||||
// 运行结束
|
||||
s.PausedTasks[task.ID] = task
|
||||
delete(s.ActiveProcesses, task.ID)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user