-
Notifications
You must be signed in to change notification settings - Fork 0
/
Scheduler.go
200 lines (178 loc) · 6.11 KB
/
Scheduler.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
package worker
import (
"fmt"
"github.com/staryjie/crontab/common"
"strings"
"time"
)
// 任务调度
type Scheduler struct {
jobEventChan chan *common.JobEvent // Etcd任务事件队列
jobPlanTable map[string]*common.JobSchedulePlan // 任务调度计划表
jobExecutingTable map[string]*common.JobExecuteInfo // 任务执行表
jobResultChan chan *common.JobExecuteResult // 任务执行结果队列
}
var (
G_scheduler *Scheduler
)
// 处理任务事件
func (scheduler *Scheduler) handlerJobEvent(jobEvent *common.JobEvent) {
var (
jobSchedulePlan *common.JobSchedulePlan
jobExecuteInfo *common.JobExecuteInfo
jobExecuting bool
jobExisted bool
err error
)
switch jobEvent.EventType {
case common.JOB_EVENT_SAVE: // 更新事件
if jobSchedulePlan, err = common.BuildJobSchedulePlan(jobEvent.Job); err != nil {
return
}
// 加入到任务调度计划表
scheduler.jobPlanTable[jobEvent.Job.Name] = jobSchedulePlan
case common.JOV_EVENT_DELETE: // 删除事件
if jobSchedulePlan, jobExisted = scheduler.jobPlanTable[jobEvent.Job.Name]; jobExisted {
delete(scheduler.jobPlanTable, jobEvent.Job.Name)
}
case common.JOB_EVENT_KILL: // 强杀任务事件
// 取消Command执行
// 判断任务是否在执行中
if jobExecuteInfo, jobExecuting = scheduler.jobExecutingTable[jobEvent.Job.Name]; jobExecuting {
jobExecuteInfo.CancelFunc() // 取消执行
}
}
}
// 尝试执行任务
func (scheduler *Scheduler) TryStartJob(jobPlan *common.JobSchedulePlan) {
// 调度和执行时两件事
var (
jobExecuteInfo *common.JobExecuteInfo
jobExecuting bool
)
// 任务执行可能要很久,但是调度很频繁,比如1分钟调度60次,单次执行要1分钟,那么只有一次能够执行,防止并发执行
// 如果任务正在执行,则跳过本次调度
if jobExecuteInfo, jobExecuting = scheduler.jobExecutingTable[jobPlan.Job.Name]; jobExecuting {
fmt.Println("任务正在执行中,跳过本次调度", jobPlan.Job.Name)
return
}
// 构建执行状态信息
jobExecuteInfo = common.BuildJobExecuteInfo(jobPlan)
// 保存执行状态
scheduler.jobExecutingTable[jobPlan.Job.Name] = jobExecuteInfo
// 执行任务
fmt.Println("执行任务:", jobExecuteInfo.Job.Name, jobExecuteInfo.PlanTime, jobExecuteInfo.RealTime)
// TODO: 执行任务,执行完成从执行表中删除该任务
G_executor.ExecuteJob(jobExecuteInfo)
}
// 重新计算任务调度状态
func (scheduler *Scheduler) TrySchedule() (scheduleAfer time.Duration) {
var (
jobPlan *common.JobSchedulePlan
now time.Time
nearTime *time.Time
)
// 如果任务表为空,睡眠1秒
if len(scheduler.jobPlanTable) == 0 {
scheduleAfer = 1 * time.Second
return
}
// 1.遍历所有任务
now = time.Now()
for _, jobPlan = range scheduler.jobPlanTable {
if jobPlan.NextTime.Before(now) || jobPlan.NextTime.Equal(now) {
// 2.过期的任务马上执行
// 尝试执行任务 // 上一个任务可能还在执行中
scheduler.TryStartJob(jobPlan)
fmt.Println(time.Now().Format("2006-01-02 15:04:05"), "执行任务:", jobPlan.Job.Name)
// 更新下一次调度时间
jobPlan.NextTime = jobPlan.Expr.Next(now)
}
// 3.计算最近一个要过期的任务,然后精确Sleep
if nearTime == nil || jobPlan.NextTime.Before(*nearTime) {
nearTime = &jobPlan.NextTime
}
}
// 精准睡眠
scheduleAfer = (*nearTime).Sub(now)
return
}
// 处理任务执行结果
func (scheduler *Scheduler) handlerJobResult(jobResult *common.JobExecuteResult) {
var (
jobLog *common.JobLog
)
// 删除任务执行表中的该任务
delete(scheduler.jobExecutingTable, jobResult.ExecuteInfo.Job.Name)
// 生成任务执行日志
if jobResult.Err != common.ERR_LOCK_ALREADY_REQUIRED {
jobLog = &common.JobLog{
JobName: jobResult.ExecuteInfo.Job.Name,
Command: jobResult.ExecuteInfo.Job.Command,
Output: string(jobResult.OutPut),
PlanTime: jobResult.ExecuteInfo.PlanTime.UnixNano() / 1000 / 1000,
ScheduleTime: jobResult.ExecuteInfo.RealTime.UnixNano() / 1000 / 1000,
StartTime: jobResult.StartTime.UnixNano() / 1000 / 1000,
EndTime: jobResult.EndTime.UnixNano() / 1000 / 1000,
}
if jobResult.Err != nil {
jobLog.Err = jobResult.Err.Error()
} else {
jobLog.Err = ""
}
// 将日志推送给MongoDB
G_logSink.Append(jobLog)
}
if jobResult.Err != nil {
fmt.Println("任务执行异常", jobResult.Err.Error())
} else {
fmt.Println("任务执行完成", jobResult.ExecuteInfo.Job.Name, strings.TrimSpace(string(jobResult.OutPut)), jobResult.Err)
}
}
// 调度协程
func (scheduler *Scheduler) scheduleLoop() {
var (
jobEvent *common.JobEvent
scheduleAfter time.Duration
scheduleTimer *time.Timer
jobResult *common.JobExecuteResult
)
// 初始化计算任务调度状态执行任务
scheduleAfter = scheduler.TrySchedule()
// 调度的延迟定时器
scheduleTimer = time.NewTimer(scheduleAfter)
// 定时任务
for {
select {
case jobEvent = <-scheduler.jobEventChan: // 监听任务变化事件
// 对内存中的任务列表做增删改查
scheduler.handlerJobEvent(jobEvent)
case <-scheduleTimer.C: // 最近的任务到期
case jobResult = <-scheduler.jobResultChan: // 监听任务执行结果
scheduler.handlerJobResult(jobResult) // 处理任务执行结果
}
// 调度一次任务
scheduleAfter = scheduler.TrySchedule()
// 重置定时器
scheduleTimer.Reset(scheduleAfter)
}
}
// 推送任务变化事件
func (scheduler *Scheduler) PushJobEvent(jobEvent *common.JobEvent) {
scheduler.jobEventChan <- jobEvent
}
// 初始化调度器
func InitScheduler() (err error) {
G_scheduler = &Scheduler{
jobEventChan: make(chan *common.JobEvent, 1000),
jobPlanTable: make(map[string]*common.JobSchedulePlan),
jobExecutingTable: make(map[string]*common.JobExecuteInfo),
jobResultChan: make(chan *common.JobExecuteResult, 1000),
}
go G_scheduler.scheduleLoop()
return
}
// 任务执行回调函数,回传任务执行结果
func (scheduler *Scheduler) PushJobResult(jobResult *common.JobExecuteResult) {
scheduler.jobResultChan <- jobResult
}