前言
golang实现定时任务很简单，只须要简单几步代码即可以完成，最近在做了几个定时任务，想研究一下它内部是怎么实现的，所以将源码过了一遍，记录和分享在此。需要的朋友可以参考以下内容，希望对大家有帮助。
关于go cron是如何使用的可以参考之前的文章：一文带你入门Go语言中定时任务库Cron的使用
Demo示例
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
|
package mainimport ( "fmt" "github.com/robfig/cron/v3")func main() { // 创建一个默认的cron对象 c := cron.New() //添加执行任务 c.AddFunc("30 * * * *", func() { fmt.Println("Every hour on the half hour") }) c.AddFunc("@hourly", func() { fmt.Println("Every hour, starting an hour from now") }) c.AddFunc("@every 1h30m", func() { fmt.Println("Every hour thirty, starting an hour thirty from now") }) //开始执行任务 c.Start() select {} //阻塞} |
通过上面的示例，可以发现， cron 最常用的几个函数：
- New(): 实例化一个 cron 对象。
- Cron.AddFunc(): 向 Cron 对象中添加一个作业，接受两个参数，第一个是 cron 表达式，第二个是一个无参无返回值的函数（作业）。
- Cron.Stop(): 停止调度，Stop 之后不会再有未执行的作业被唤醒，但已经开始执行的作业不会受影响。
源码实现
在了解其整体逻辑的实现过程前，先了解两个重要的结构体Entry和Cron：
位置在/robfig/cron/cron.go。
结构体 Cron 和 Entry
Cron主要负责维护所有的任务数据，调用相关的func时间指定，可以启动、停止任务等；Entry是对添加到 Cron 中的任务的封装，每个 Entry 有一个 ID，除此之外，Entry 里保存了这个任务上次运行的时间和下次运行的时间。具体代码实现如下：
|
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
|
// Entry 数据结构，每一个被调度实体一个type Entry struct { // 唯一id，用于查询和删除 ID EntryID // 本Entry的调度时间，不是绝对时间，在生成entry时会计算出来 Schedule Schedule // 本entry下次需要执行的绝对时间，会一直被更新 // 被封装的含义是Job可以多层嵌套，可以实现基于需要执行Job的额外处理 // 比如抓取Job异常、如果Job没有返回下一个时间点的Job是还是继续执行还是delay Next time.Time // 上一次被执行时间，主要用来查询 Prev time.Time // WrappedJob 是真实执行的Job实体 WrappedJob Job // Job 主要给用户查询 Job Job}// Cron保持任意数量的任务的轨道，调用相关的func时间表指定。它可以被启动，停止，可运行的同时进行检查。type Cron struct { entries []*Entry // 保存了所有加入到 Cron 的任务 // chain 用来定义entry里的warppedJob使用什么逻辑（e.g. skipIfLastRunning） // 即一个cron里所有entry只有一个封装逻辑 chain Chain stop chan struct{} // 停止整个cron的channel add chan *Entry // 增加一个entry的channel remove chan EntryID // 移除一个entry的channel snapshot chan chan []Entry // 获取entry整体快照的channel running bool // 代表是否已经在执行，是cron为使用者提供的动态修改entry的接口准备的 logger Logger // 封装golang的log包 runningMu sync.Mutex // 用来修改运行中的cron数据，比如增加entry，移除entry location *time.Location // 地理位置 parser ScheduleParser // 对时间格式的解析，为interface, 可以定制自己的时间规则。 nextID EntryID // entry的全局ID，新增一个entry就加1 jobWaiter sync.WaitGroup // run job时会进行add(1)， job 结束会done()，stop整个cron，以此保证所有job都能退出} |
New()实现
cron.go中的New()方法用来创建并返回一个Cron对象指针，其实现如下：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
func New(opts ...Option) *Cron { c := &Cron{ entries: nil, chain: NewChain(), add: make(chan *Entry), stop: make(chan struct{}), snapshot: make(chan chan []Entry), remove: make(chan EntryID), running: false, runningMu: sync.Mutex{}, logger: DefaultLogger, location: time.Local, parser: standardParser, } for _, opt := range opts { opt(c) } return c} |
AddFunc()实现
AddFunc() 用于向Corn中添加一个任务，AddFunc()中将func包装成 Job 类型然后调用AddJob()，AddFunc() 相较于 AddJob() 帮用户省去了包装成 Job 类型的一步，在 AddJob() 中，调用了 standardParser.Parse() 将 cron 表达式解释成了 schedule 类型，最终，他们调用了 Schedule() 方法；其代码实现如下：
|
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
|
func (c *Cron) AddFunc(spec string, cmd func()) (EntryID, error) { return c.AddJob(spec, FuncJob(cmd)) //包装成job类型然后调用AddJob()方法}func (c *Cron) AddJob(spec string, cmd Job) (EntryID, error) { schedule, err := c.parser.Parse(spec) //将cron表达式解析成schedule类型 if err != nil { return 0, err } return c.Schedule(schedule, cmd), nil //调用Schedule()}func (c *Cron) Schedule(schedule Schedule, cmd Job) EntryID { c.runningMu.Lock() //为了保证线程安全，加锁 defer c.runningMu.Unlock() c.nextID++ //下一EntryID entry := &Entry{ ID: c.nextID, Schedule: schedule, WrappedJob: c.chain.Then(cmd), Job: cmd, } // Cron是否处于运行状态 if !c.running { c.entries = append(c.entries, entry) // 追加到entries列表中 } else { c.add <- entry // 发送到Cron的add chan } return entry.ID} |
Schedule()这个方法负责创建 Entry 结构体，并把它追加到 Cron 的 entries 列表中，如果 Cron 已经处于运行状态，会将这个创建好的 entry 发送到 Cron的 add chan 中，在run()中会处理这种情况。
Start()实现
Start() 用于开始执行 Cron,其代码实现如下：
|
1
2
3
4
5
6
7
8
9
|
func (c *Cron) Start() { c.runningMu.Lock() // 获取锁 defer c.runningMu.Unlock() if c.running { return } c.running = true // 将 c.running 置为 true 表示 cron 已经在运行中了 go c.run() //开启一个 goroutine 执行 c.run()} |
通过上面的代码，可以看到主要干了这么几件事：
- 获取锁，保证线程安全。
-
判断
cron是否已经在运行中，如果是则直接返回，否则将c.running置为true表示cron已经在运行中了。 -
开启一个
goroutine执行c.run()。
Run()实现
Run()是整个cron的一个核心，它负责处理cron开始执行后的大部分事情， run中会一直轮循c.entries中的entry, 如果一个entry 允许执行了，就会开启单独的goroutine去执行这个任务。
|
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
|
// run the scheduler.. this is private just due to the need to synchronize// access to the 'running' state variable.func (c *Cron) run() { c.logger.Info("start") // Figure out the next activation times for each entry. now := c.now() for _, entry := range c.entries { entry.Next = entry.Schedule.Next(now) c.logger.Info("schedule", "now", now, "entry", entry.ID, "next", entry.Next) } for { // Determine the next entry to run. // 将定时任务执行时间进行排序,最近最早执行的放在前面 sort.Sort(byTime(c.entries)) var timer *time.Timer if len(c.entries) == 0 || c.entries[0].Next.IsZero() { // If there are no entries yet, just sleep - it still handles new entries // and stop requests. timer = time.NewTimer(100000 * time.Hour) } else { // 生成一个定时器,距离最近的任务时间到时 触发定时器的channel,发送通知 timer = time.NewTimer(c.entries[0].Next.Sub(now)) } for { select { // 定时时间到了,执行定时任务,并设置下次执行的时刻 case now = <-timer.C: now = now.In(c.location) c.logger.Info("wake", "now", now) // Run every entry whose next time was less than now //对每个定时任务尝试执行 for _, e := range c.entries { if e.Next.After(now) || e.Next.IsZero() { break } c.startJob(e.WrappedJob) e.Prev = e.Next e.Next = e.Schedule.Next(now) c.logger.Info("run", "now", now, "entry", e.ID, "next", e.Next) } //新增的定时任务添加到 任务列表中 case newEntry := <-c.add: timer.Stop() now = c.now() newEntry.Next = newEntry.Schedule.Next(now) c.entries = append(c.entries, newEntry) c.logger.Info("added", "now", now, "entry", newEntry.ID, "next", newEntry.Next) //获取 当前所有定时任务(快照) case replyChan := <-c.snapshot: replyChan <- c.entrySnapshot() continue //停止定时任务,timer停止即可完成此功能 case <-c.stop: timer.Stop() c.logger.Info("stop") return //删除某个定时任务 case id := <-c.remove: timer.Stop() now = c.now() c.removeEntry(id) c.logger.Info("removed", "entry", id) } break } }} |
Stop()实现
Stop() 用来停止Cron的运行，但已经在执行中的作业是不会被打断的，也就是从执行 Stop() 之后，不会再有新的任务被调度：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
|
func (c *Cron) Stop() context.Context { c.runningMu.Lock() defer c.runningMu.Unlock() if c.running { c.stop <- struct{}{} // 会发出一个 stop 信号 c.running = false } ctx, cancel := context.WithCancel(context.Background()) go func() { // 等待所有已经在执行的任务执行完毕 c.jobWaiter.Wait() // 会发出一个 cancelCtx.Done() 信号 cancel() }() return ctx} |
Remove()实现
Remove() 用于移除一个任务：
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
func (c *Cron) Remove(id EntryID) { c.runningMu.Lock() defer c.runningMu.Unlock() if c.running { c.remove <- id // 会发出一个 remove 信号 } else { c.removeEntry(id) }}func (c *Cron) removeEntry(id EntryID) { var entries []*Entry for _, e := range c.entries { if e.ID != id { entries = append(entries, e) } } c.entries = entries} |
小结
到此这篇关于Golang Cron 定时任务的内部实现的文章就介绍到这了, 其中重点如下：
在Go Cron内部维护了两个结构体Cron和Entry，用于维护任务数据，cron.Start()执行后，cron的后台程序c.Run()就开始执行了，Run()是整个cron的一个核心，它负责处理cron开始执行后的大部分事情， run中会一直轮循c.entries中的entry, 每个entry都包含自己下一次执行的绝对时间，如果一个entry 允许执行了，就会开启单独的goroutine去执行这个任务。
以上就是通过源码分析Golang cron的实现原理的详细内容，更多关于Golang cron的资料请关注服务器之家其它相关文章！
原文链接：https://juejin.cn/post/7155769581255000101
