Go组件学习——cron定时器

梦想是做不了贪官做老赖 發表於 2019-7-28 18:54:00

<h3 id="1">1 前言</h3>
　　转到Go已经将近三个月，写业务代码又找到了属于Go的条件反射了。
　　后置声明和多参数返回这些Go风格代码写起来也不会那么蹩脚，甚至还有点小适应~
　　反而，前几天在写Java的时候，发现Java怎么启动这么慢，Java怎么能够容忍这些用不到的代码还理直气壮的躺在那……等等，这些话在哪听过类似的？？？
　　“Go为什么要后置声明，多别扭啊”
　　“Go里面为啥要定义这么多的struct，看的头晕”
　　……
　　其实，没有最好的语言，只有最适合的。
 
　　前面《Go语言学习》系列主要介绍了一些Go的基础知识和相较于Java的一些新特性。后续如果有相关的体会和新的还会继续更新。
　　从这篇开始，开始学习Go的一些工具类库和开源组件，希望在学习这些优秀的开源项目过程中，更深入的了解Go，发现Go的威力。
 
<h3 id="2cron">2 cron简介</h3>
　　robfig/cron是一个第三方开源的任务调度库，也就是我们平时说的定时任务。
　　Github：https://github.com/robfig/cron
　　官方文档：https://godoc.org/github.com/robfig/cron
<h3> </h3>
<h3 id="3cron">3 cron如何使用</h3>
1、新建文件cron-demo.go
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">package main

import (
"fmt"
"github.com/robfig/cron"
"time"
)

func main() {
c := cron.New()
c.AddFunc("*/3 * * * * *", func() {
fmt.Println("every 3 seconds executing")
})

go c.Start()
defer c.Stop()

select {
case <-time.After(time.Second * 10):
return
}
}</pre>
</div>
<ul class=" list-paddingleft-2">
<li>
cron.New创建一个定时器管理器
</li>
<li>
c.AddFunc添加一个定时任务，第一个参数是cron时间表达式，第二个参数是要触发执行的函数
</li>
<li>
go c.Start()新启一个协程，运行定时任务
</li>
<li>
c.Stop是等待停止信号结束任务
</li>
</ul>
 
2、在cron-demo.go文件下执行go build
本项目采用go mod进行包管理，所以执行go build命令后，会在go.mod文件中生成对应的依赖版本如图所示
<img src="https://img2018.cnblogs.com/blog/619240/201907/619240-20190728184355538-735301169.png" alt="">
 
3、运行cron-demo.go
<img src="https://img2018.cnblogs.com/blog/619240/201907/619240-20190728184407571-2119990512.gif" alt="">
　　可以看出每3秒执行一次，直到10秒后过期退出进程，任务结束。
　　代码参见项目：go-demo项目（https://github.com/DMinerJackie/go-demo/tree/master/main/src/cron）
 
看上去这个任务调度还是蛮好用的，那么具体是如何实现的呢，看了下源码，也是非常的短小精悍，目录结构如下。
<img src="https://img2018.cnblogs.com/blog/619240/201907/619240-20190728184419741-560724681.png" alt="">
下面通过几个问题一起看下cron是如何实现任务调度。
<h3> </h3>
<h3 id="4cron">4 cron如何解析任务表达式</h3>
　　上例我们看到添加“*/3 * * * * *”这样的表达式，就能实现每3秒执行一次。
　　显然，这个表达式只是对人友好的一种约定表达形式，要真正在指定时间执行任务，cron肯定是要读取并解析这个c表达式，转化为具体的时间再执行。
　　那我们来看看，这个具体是如何执行的。
　　进入AddFunc函数实现
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">// AddFunc adds a func to the Cron to be run on the given schedule.
func (c *Cron) AddFunc(spec string, cmd func()) error {
return c.AddJob(spec, FuncJob(cmd))
}
</pre>
</div>
　　
　　这只是套了个壳，具体还要进入AddJob函数
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">// AddJob adds a Job to the Cron to be run on the given schedule.
func (c *Cron) AddJob(spec string, cmd Job) error {
schedule, err := Parse(spec)
if err != nil {
return err
}
c.Schedule(schedule, cmd)
return nil
}
</pre>
</div>
　　
　　该函数第一行就是解析cron表达式，顺藤摸瓜，我们看到具体实现如下
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">// Parse returns a new crontab schedule representing the given spec.
// It returns a descriptive error if the spec is not valid.
// It accepts crontab specs and features configured by NewParser.
func (p Parser) Parse(spec string) (Schedule, error) {
if len(spec) == 0 {
return nil, fmt.Errorf("Empty spec string")
}
if spec == '@' && p.options&Descriptor > 0 {
return parseDescriptor(spec)
}

// Figure out how many fields we need
max := 0
for _, place := range places {
if p.options&place > 0 {
max++
}
}
min := max - p.optionals

// Split fields on whitespace
fields := strings.Fields(spec) // 使用空白符拆分cron表达式

// Validate number of fields
if count := len(fields); count < min || count > max {
if min == max {
return nil, fmt.Errorf("Expected exactly %d fields, found %d: %s", min, count, spec)
}
return nil, fmt.Errorf("Expected %d to %d fields, found %d: %s", min, max, count, spec)
}

// Fill in missing fields
fields = expandFields(fields, p.options)

var err error
field := func(field string, r bounds) uint64 { // 抽象出filed函数，方便下面调用
if err != nil {
return 0
}
var bits uint64
bits, err = getField(field, r)
return bits
}

var (
second = field(fields, seconds)
minute = field(fields, minutes)
hour = field(fields, hours)
dayofmonth = field(fields, dom)
month = field(fields, months)
dayofweek= field(fields, dow)
)
if err != nil {
return nil, err
}

return &SpecSchedule{
Second: second,
Minute: minute,
Hour: hour,
Dom: dayofmonth,
Month:month,
Dow: dayofweek,
}, nil
}
</pre>
</div>
　　
　　该函数主要是将cron表达式映射为“Second, Minute, Hour, Dom, Month, Dow”6个时间维度的结构体SpecSchedule。
　　SpecSchedule是实现了方法“Next(time.Time) time.Time”的结构体，而“Next(time.Time) time.Time”是定义在Schedule接口中的
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">// The Schedule describes a job's duty cycle.
type Schedule interface {
// Return the next activation time, later than the given time.
// Next is invoked initially, and then each time the job is run.
Next(time.Time) time.Time
}</pre>
</div>
　　所以，最终可以理解是将cron解析后转换为下一次要执行的时刻，等待执行。
<h3> </h3>
<h3 id="5cron">5 cron如何执行任务</h3>
　　我们知道通过parser.go可以将人很好理解的表达式转换为cron可以读懂的要执行的时间。
　　有了要执行的时间点，那么cron具体是如何执行这些任务的呢？
　　我们看下Start函数的具体实现
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">// Start the cron scheduler in its own go-routine, or no-op if already started.
func (c *Cron) Start() {
if c.running {
return
}
c.running = true
go c.run()
}</pre>
</div>
　　这里会通过判定Cron的running字段是否在运行来巨额听是否要启动任务。
　　显然这里running是false，因为在调用c.New初始化的时候running被设置为false。
　　所以，这里新启一个协程用于执行定时任务，再次顺藤摸瓜，我们看到run函数的实现
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">// Run the scheduler. this is private just due to the need to synchronize
// access to the 'running' state variable.
func (c *Cron) run() {
// Figure out the next activation times for each entry.
now := c.now()
for _, entry := range c.entries {
entry.Next = entry.Schedule.Next(now)
}

for {
// Determine the next entry to run.
sort.Sort(byTime(c.entries))

var timer *time.Timer
if len(c.entries) == 0 || c.entries.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 {
timer = time.NewTimer(c.entries.Next.Sub(now)) // 否则新建一个距离现在到下一个要触发执行的Timer
}

for {
select {
case now = <-timer.C: // 触发时间到，执行任务
now = now.In(c.location)
// Run every entry whose next time was less than now
for _, e := range c.entries {
if e.Next.After(now) || e.Next.IsZero() {
break
}
go c.runWithRecovery(e.Job)
e.Prev = e.Next
e.Next = e.Schedule.Next(now)
}

case newEntry := <-c.add: // 添加任务
timer.Stop()
now = c.now()
newEntry.Next = newEntry.Schedule.Next(now)
c.entries = append(c.entries, newEntry)

case <-c.snapshot: // 调用c.Entries()返回一个现有任务列表的snapshot
c.snapshot <- c.entrySnapshot()
continue

case <-c.stop: // 任务结束，退出
timer.Stop()
return
}

break
}
}
}</pre>
</div>
<ul class=" list-paddingleft-2">
<li>
进入该函数，首先遍历所以任务，找到所有任务下一个要执行的时间。
</li>
<li>
然后进入外层for循环，对于各个任务按照执行时间进行排序，保证离当前时间最近的先执行。
</li>
<li>
再对任务列表进行判定，是否有任务如果没有，则休眠，否则初始化一个timer。
</li>
</ul>
 
里层的for循环才是重头戏，下面主要分析这个for循环里面的任务加入和执行。
在此之前，需要了解下go标准库的timer
timer用于指定在某个时间间隔后，调用函数或者表达式。
使用NewTimer就可以创建一个Timer，在指定时间间隔到达后，可以通过<-timer.C接收值。
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">package main

import (
"fmt"
"time"
)

func main() {
timer1 := time.NewTimer(2 * time.Second)

<-timer1.C
fmt.Println("Timer 1 expired")

timer2 := time.NewTimer(time.Second)
go func() {
<-timer2.C
fmt.Println("Timer 2 expired")
}()

stop2 := timer2.Stop()
if stop2 {
fmt.Println("Timer 2 stopped")
}
}</pre>
</div>
　　执行结果为
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">Timer 1 expired
Timer 2 stopped
</pre>
</div>
　　timer1表示2秒后到期，在此之前都是阻塞状态，2秒后<-timer1.C接收到信号，执行下面的打印语句。
　　timer2表示1秒后到期，但是中途被Stop掉了，相当于清除了定时功能。
 
　　有了这个背景之后，我们再来看run函数的里层for循环。
　　接收到c.add信道
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">case newEntry := <-c.add: // 添加任务
timer.Stop()
now = c.now()
newEntry.Next = newEntry.Schedule.Next(now)
c.entries = append(c.entries, newEntry)</pre>
</div>
　　将timer停掉，清除设置的定时功能，并以当前时间点为起点，设置添加任务的下一次执行时间，并添加到entries任务队列中。
 
　　接收到timer.C信道
 
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">case now = <-timer.C: // 触发时间到，执行任务
now = now.In(c.location)
// Run every entry whose next time was less than now
for _, e := range c.entries {
if e.Next.After(now) || e.Next.IsZero() {
break
}
go c.runWithRecovery(e.Job)
e.Prev = e.Next
e.Next = e.Schedule.Next(now)
}</pre>
</div>
 
　　当定任务到点后，time.C就会接收到值，并新开协程执行真正需要执行的Job，之后再更新下一个要执行的任务列表。
　　我们进入runWithRecovery函数，该函数从函数名就可以看出，即使出现panic也可以重新recovery，保证其他任务不受影响。
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">func (c *Cron) runWithRecovery(j Job) {
defer func() {
if r := recover(); r != nil {
const size = 64 << 10
buf := make([]byte, size)
buf = buf[:runtime.Stack(buf, false)]
c.logf("cron: panic running job: %v\n%s", r, buf)
}
}()
j.Run()
}</pre>
</div>
追根溯源，我们发现真正执行Job的是j.Run()的执行。进入这个Run函数的实现，我们看到
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">func (f FuncJob) Run() { f() }</pre>
</div>
没错，我们要执行的任务一直从AddFunc一直往下传递，直到这里，我们通过调用Run函数，将包装的FuncJob类型的函数通过f()的形式进行执行。
这里说的可能比较模糊，举个例子，Go里面的闭包定义
<div class="cnblogs_Highlighter">
<pre class="brush:go;gutter:true;">func () {
fmt.Println("test")
}()</pre>
</div>
如果这里定义后面没有"()"该函数就不会执行，所以结合这个看上面的定时任务是如何执行就更容易理解了。
<h3> </h3>
<h3 id="6">6 代码阅读体会</h3>
1、channel的奥妙
　　通过channel可以让感知变得轻而易举，比如timer.C就像是时间到了，自然会有人来敲门告诉你。而不需要我们自己主动去获取是否到期了。
2、常用类库的使用
　　比如在parser里面我们看到了"fields := strings.Fields(spec)"，在日常开发中，我们可以灵活使用这些API，避免自己造轮子的情况。
3、多思考
　　之前做Java的时候，更多的是沉浸在各种工具和框架的使用，对于这些工具和框架的实现关注的不多。比如从Quartz到Spring Job，我们需要更新的是越来越好用的定时任务工具，而底层的实现升级Spring都帮我们考虑好了。这种对业务对项目有友好的，可以快速的实现业务功能开发，但是对于开发者并不友好，友好的设计麻痹了开发者对于底层原理的深究的欲望。
 
来源：https://www.cnblogs.com/bigdataZJ/p/go-opensource-cron.html

頁: [1]

琼殿技术论坛's Archiver

Go组件学习——cron定时器