-
Notifications
You must be signed in to change notification settings - Fork 0
/
scheduler.go
246 lines (211 loc) · 6.88 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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
package worker
import (
"fmt"
"github.com/qiuqiu1999/crontab/common"
"math/rand"
"os/exec"
"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) handleJobEvent(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.JOB_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() // 触发command杀死shell子进程, 任务得到退出
}
}
}
// 尝试执行任务
func (scheduler *Scheduler) TryStartJob(jobPlan *common.JobSchedulePlan) {
// 调度 和 执行 是2件事情
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)
scheduler.ExecuteJob(jobExecuteInfo)
}
// 重新计算任务调度状态
func (scheduler *Scheduler) TrySchedule() time.Duration {
var (
jobPlan *common.JobSchedulePlan
now time.Time
nearTime *time.Time
scheduleAfter time.Duration
)
// 如果任务表为空话,随便睡眠多久
if len(scheduler.jobPlanTable) == 0 {
scheduleAfter = 1 * time.Second
return scheduleAfter
}
// 当前时间
now = time.Now()
// 遍历所有任务
for _, jobPlan = range scheduler.jobPlanTable {
if jobPlan.NextTime.Before(now) || jobPlan.NextTime.Equal(now) {
scheduler.TryStartJob(jobPlan)
jobPlan.NextTime = jobPlan.Expr.Next(now) // 更新下次执行时间
}
// 统计最近一个要过期的任务时间
if nearTime == nil || jobPlan.NextTime.Before(*nearTime) {
nearTime = &jobPlan.NextTime
}
}
// 下次调度间隔(最近要执行的任务调度时间 - 当前时间)
scheduleAfter = (*nearTime).Sub(now)
return scheduleAfter
}
// 处理任务结果
func (scheduler *Scheduler) handleJobResult(result *common.JobExecuteResult) {
var (
jobLog *common.JobLog
)
// 删除执行状态
delete(scheduler.jobExecutingTable, result.ExecuteInfo.Job.Name)
// 生成执行日志
if result.Err != common.ERR_LOCK_ALREADY_REQUIRED {
jobLog = &common.JobLog{
JobName: result.ExecuteInfo.Job.Name,
Command: result.ExecuteInfo.Job.Command,
Output: string(result.Output),
PlanTime: result.ExecuteInfo.PlanTime.UnixNano() / 1000 / 1000,
ScheduleTime: result.ExecuteInfo.RealTime.UnixNano() / 1000 / 1000,
StartTime: result.StartTime.UnixNano() / 1000 / 1000,
EndTime: result.EndTime.UnixNano() / 1000 / 1000,
}
if result.Err != nil {
jobLog.Err = result.Err.Error()
} else {
jobLog.Err = ""
}
G_logSink.Append(jobLog)
}
// fmt.Println("任务执行完成:", result.ExecuteInfo.Job.Name, string(result.Output), result.Err)
}
// 调度协程
func (scheduler *Scheduler) scheduleLoop() {
var (
jobEvent *common.JobEvent
scheduleAfter time.Duration
scheduleTimer *time.Timer
jobResult *common.JobExecuteResult
)
// 初始化一次(1秒)
scheduleAfter = scheduler.TrySchedule()
// 调度的延迟定时器
scheduleTimer = time.NewTimer(scheduleAfter)
// 定时任务common.Job
for {
select {
case jobEvent = <-scheduler.jobEventChan: //监听任务变化事件
// 对内存中维护的任务列表做增删改查
scheduler.handleJobEvent(jobEvent)
case <-scheduleTimer.C: // 最近的任务到期了
case jobResult = <-scheduler.jobResultChan: // 监听任务执行结果
scheduler.handleJobResult(jobResult)
}
// 调度一次任务
scheduleAfter = scheduler.TrySchedule()
// 重置调度间隔
scheduleTimer.Reset(scheduleAfter)
}
}
// 推送任务变化事件
func (scheduler *Scheduler) PushJobEvent(jobEvent *common.JobEvent) {
scheduler.jobEventChan <- jobEvent
}
// 初始化调度器
func InitScheduler() {
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()
}
// 回传任务执行结果
func (scheduler *Scheduler) PushJobResult(jobResult *common.JobExecuteResult) {
scheduler.jobResultChan <- jobResult
}
func (scheduler *Scheduler) ExecuteJob(info *common.JobExecuteInfo) {
go func() {
var (
cmd *exec.Cmd
err error
output []byte
result *common.JobExecuteResult
jobLock *JobLock
)
// 任务结果
result = &common.JobExecuteResult{
ExecuteInfo: info,
Output: make([]byte, 0),
}
// 初始化分布式锁
jobLock = G_jobMgr.CreateJobLock(info.Job.Name)
// 记录任务开始时间
result.StartTime = time.Now()
// 上锁
// 随机睡眠(0~1s)
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
err = jobLock.TryLock()
defer jobLock.Unlock()
if err != nil { // 上锁失败
result.Err = err
result.EndTime = time.Now()
} else {
// 上锁成功后,重置任务启动时间
result.StartTime = time.Now()
// 执行shell命令
cmd = exec.CommandContext(info.CancelCtx, "/bin/bash", "-c", info.Job.Command)
// 执行并捕获输出
output, err = cmd.CombinedOutput()
// 记录任务结束时间
result.EndTime = time.Now()
result.Output = output
result.Err = err
}
// 任务执行完成后,把执行的结果返回给Scheduler,Scheduler会从executingTable中删除掉执行记录
scheduler.PushJobResult(result)
}()
}