外观
Go语言中轻量且易用的定时任务库 – 原理解析
本文为稀土掘金技术社区首发签约文章，30天内禁止转载，30天后未获授权禁止转载，侵权必究！
在现代的软件开发中，定时任务是一项非常常见的需求。无论是周期性的数据备份、定时发送邮件、定时执行清理任务还是其他自动化操作，定时任务都是提高系统效率和可靠性的重要工具。Go 语言作为一种快速、高效、并发的编程语言，也有自己的定时任务库，其中最受欢迎的之一就是 robfig/cron。
介绍
robfig/cron 是由 Rob Figueiredo 创建和维护的一个开源项目。它为 Go 应用程序提供了灵活、易用的定时任务调度功能，使开发人员能够轻松地在应用程序中定义和调度各种定时任务。
为什么选择 robfig/cron？
- 灵活的任务调度：
robfig/cron允许开发人员根据时间、间隔、重复次数等条件来执行任务，具有很高的灵活性。无论是每分钟、每小时、每天、每周还是其他自定义的时间间隔，都可以轻松地设置。 - 多任务支持：它支持同时运行多个任务，并提供了一套简洁的 API 来控制任务的执行。这意味着你可以在同一个程序中管理和调度多个不同的定时任务，而无需担心冲突或混乱。
- 简单易用：API 设计简洁明了，易于理解和使用，即使对于 Go 新手也很容易上手。通过简单的几行代码就可以实现定时任务的定义和调度，无需复杂的配置和学习曲线。
- 可靠性：
robfig/cron提供了一些机制来处理任务执行过程中的错误，并确保任务的可靠执行。无论是处理异常情况还是保证任务按时执行，都能够得到良好的支持和保障。
如何使用 robfig/cron？
下面是一个简单的示例代码，演示了如何使用 robfig/cron 在 Go 应用程序中执行定时任务：
package main
import (
"fmt"
"github.com/robfig/cron"
)
func main() {
c := cron.New()
// 添加一个每分钟执行一次的定时任务
c.AddFunc("* * * * *", func() {
fmt.Println("执行定时任务：每分钟执行一次")
})
// 添加一个每天凌晨执行的定时任务
c.AddFunc("0 0 * * *", func() {
fmt.Println("执行定时任务：每天凌晨执行一次")
})
// 启动定时任务
c.Start()
// 阻塞主线程，直到接收到退出信号
select {}
}在这个示例中，我们首先创建了一个 cron.Cron 实例，然后使用 AddFunc 方法向其添加了两个定时任务。第一个任务每分钟执行一次，第二个任务每天凌晨执行一次。最后，我们调用 Start 方法启动定时任务，并通过 select {} 语句使主线程进入阻塞状态，以防止程序退出。
源码分析
核心结构
Cron Struct
cron struct是整个库最为核心的结构，它负责任务的管控和调度
type Cron struct {
entries []*Entry //在cron中每一个提交都任务就是一个entry
chain Chain //Chain一组JobWrapper的集合，JobWrapper可以对任务进行自定义包装，
// 通过chain可以将多个任务进行关联执行
stop chan struct{} //stop chan
add chan *Entry //add chan
remove chan EntryID. //remove chan
snapshot chan chan []Entry //entry 集合快照
running bool //是否在运行中
logger Logger //日志接口，可以由用户自行实现
runningMu sync.Mutex //排他锁
location *time.Location //time location, 可以由用户指定
parser ScheduleParser //时间表达式 解析接口，默认提供，也可以由用户自由实现
nextID EntryID //下一个EntryID，EntryID 自增
jobWaiter sync.WaitGroup //主要用来保证，在停止cron时，需要等待正在运行的任务，运行完成后再关闭上下文
}Entry Struct
entry struct 是cron库中的最小调度单元，相当于每一个用户通过AddFunc提交的任务都会被转换成Entry实例，被cron struct进行管理与调度
type Entry struct {
ID EntryID //唯一ID
Schedule Schedule //Scheduler 是一个接口，接口中Next可以获取job下一次运行时间
Next time.Time //下一次job运行时间
// Prev is the last time this job was run, or the zero time if never.
Prev time.Time //最后一次job运行时间
// WrappedJob is the thing to run when the Schedule is activated.
WrappedJob Job //调度时真正运行的job接口，job接口中的run func 就是用户需要被调度的run func
// Job is the thing that was submitted to cron.
// It is kept around so that user code that needs to get at the job later,
// e.g. via Entries() can do so.
Job Job // 提交的任务
}核心函数
Cron.New
通过cron.New 创建cron， New func 通过option 方式 接收参数
例如，可以指定任务的执行策略,例如指定为DelayIfStillRunning
DelayStillRunning 代表当定时任务逾期时，下一次调度要一直等待本次任务调度完毕后再执行。
cron.New(cron.WithChain(cron.DelayIfStillRunning(logger)))这样就会在cron 中的chain 集合中增加一个job wrapper。job wrapper 是一个接口，可以由用户自由实现，job wrapper中 主要是对在真正用户实现的run接口 之上在包装一层。类似于web 框架中的middleware.
type JobWrapper func(Job) Job看一下DelayIfStillRunning 是如何实现的
func DelayIfStillRunning(logger Logger) JobWrapper {
return func(j Job) Job {
var mu sync.Mutex
return FuncJob(func() {
start := time.Now()
mu.Lock()
defer mu.Unlock()
if dur := time.Since(start); dur > time.Minute {
logger.Info("delay", "duration", dur)
}
j.Run()
})
}
}在FuncJob 其实就是一个实现Run func的Job， 每次任务调度开始时先获取当前时间，然后加锁，如果上一次任务一直没有执行完，那么这个锁就一直会这里阻塞住，达成当上一次任务没有运行完成下一次调度会一直延迟等待。如果阻塞等待超过一分钟会打印日志记录一下。
Cron.AddJob
此函数主要作用是添加一个任务，例如：添加每30S运行一次的任务
cron.AddJob("30 * * * * ?",&job1)源码：
首先会根据用户设置的cron表达式,解析出schedule，schedule是一个interface, 用户可以自定义解析器，如果不指定则使用默认的SpecSchedule。 SpecSchedule就是使用默认cron表达式对用户传递的表达式进行解析。然后调用Schedule func将job 添加到Cron中
func (c *Cron) AddJob(spec string, cmd Job) (EntryID, error) {
schedule, err := c.parser.Parse(spec)
if err != nil {
return 0, err
}
return c.Schedule(schedule, cmd), nil
}- 真正将job 添加到cron中，这里防止并发问题会先加写锁
- 将nextID自增加1，entry是唯一的用户确认任务的唯一性，nextID会返回给用户，方便用户对任务进行管控，例如删除
- 生成Entry实例，并将当前任务添加到chain中
- 判断cron是否正在调度中，如果没有运行中 直接追加到entries就可以，等待cron被正式启动后任务就会按照用户设置的时间自动调度
- 如果cron已经在运行中了，则需要把当前entry(任务)往add 的chan中添加，触发add事件
- 返回给用户entryID=nextID
func (c *Cron) Schedule(schedule Schedule, cmd Job) EntryID {
c.runningMu.Lock()
defer c.runningMu.Unlock()
c.nextID++
entry := &Entry{
ID: c.nextID,
Schedule: schedule,
WrappedJob: c.chain.Then(cmd),
Job: cmd,
}
if !c.running {
c.entries = append(c.entries, entry)
} else {
c.add <- entry
}
return entry.ID
}上一步构建entry时, 注意entry中的wrappedJob 是通过c.chain.Then(cmd)得到的。看一下Then，这个func 挺有意思的 遍历所有wrappers，wrappers就是cron.New(cron.WithChan(…JobWrappers)) 中的JobWrappers，也就是上面举例说的 DelayIfStillRunning。
then中遍历所有的wrappers 然后得到wrappers中的 Job， 通过一个倒序遍历后，最后将最第一个JobWrapper中的Job 返回，设置到Entry中。
func (c Chain) Then(j Job) Job {
for i := range c.wrappers {
j = c.wrappers[len(c.wrappers)-i-1](j)
}
return j
}Cron.Start
启动cron 调度， 加锁防止并发，如果已经启动过了 则直接返回，否则开启goroutine调用 run()
func (c *Cron) Start() {
c.runningMu.Lock()
defer c.runningMu.Unlock()
if c.running {
return
}
c.running = true
go c.run()
}最核心的地方，看一下cron真正的run函数，是如何调度用户注册的任务的，这里是核心源码，下面会拆解进行详细描述。
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 {
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
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
}
}
}可以结合这个图来对下面的详细描述，进行理解。
- 首选得到当前时间，然后根据当前时间对设置每一个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)
}- cron启动后 遍历所有entry,然后循环将entry设置下一次调用的时间。开启第一层循环，在循环中对entry 按照下一次调用时间由近到远排序
sort.Sort(byTime(c.entries))- 创建timer 定时调度,timer 的触发时间 就是第一个将要被调度的任务的时间减去当前时间得到的时间差（这里就是整体所有添加到cron的任务，第一个任务要被调度的时间了）
timer = time.NewTimer(c.entries[0].Next.Sub(now))- 开启第二层循环，第二层循环中监听了好几种chan
监听timer.C的chan
如果Timer时间到了则触发调用。调用的逻辑是遍历entrys，如果因为entrys是有序的，所以timer.C的chan接收到了消息，则说明一定是有任务到执行时间了，执行startJob调用job任务即可(此处的job是jobwrapper) jobwrapper 经过层层调用，最终就会调用到我们自定义的Job run func中
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)当遍历到entry的下一次执行时间大于当前时间，则说明这个entry和后面的entry都是未到时间的 所以直接跳出循环即可
for _, e := range c.entries {
if e.Next.After(now) || e.Next.IsZero() {
break
}
}当调度完成后需要设置任务的上一次调度时间和下一次调度时间（上一次调度时间主要是为了信息展示，下一次调度时间则是核心，是为了对entry进行排序然后创建timer进行触发）
c.startJob(e.WrappedJob)
e.Prev = e.Next
e.Next = e.Schedule.Next(now)监听add的chan
首先停止timer，然后将当前新添加的entry 的下一次调度时间计算出来，添加到entry集合中
跳出第二层循环，停止timer是为了停止调度，因为有可能新加入的job的调度时间即将执行，所以需要重新对entry排序。
case newEntry := <-c.add:
timer.Stop()
now = c.now()
newEntry.Next = newEntry.Schedule.Next(now)
c.entries = append(c.entries, newEntry)监听snapshot的chan
shapshot 是 chan chan []entry的结构。当调用API获取entrys时，会往shanpshot 写入一个chan []Entry,然后在将entrys 副本在写入 chan []entry中。API 从 chan []entry 得到数据后返回给调用者
case replyChan := <-c.snapshot:
replyChan <- c.entrySnapshot()
continue
// entrySnapshot returns a copy of the current cron entry list.
func (c *Cron) entrySnapshot() []Entry {
var entries = make([]Entry, len(c.entries))
for i, e := range c.entries {
entries[i] = *e
}
return entries
}总结
robfig/cron 是一个功能强大、灵活、易用的定时任务库，为 Go 开发人员提供了一种方便的方式来处理定时任务。无论是简单的定时任务还是复杂的定时调度，都可以通过 robfig/cron 轻松实现。通过它可以结合分布式锁实现分布式的任务调度，如果你正在寻找一个可靠的定时任务解决方案，不妨考虑使用 robfig/cron。
原文链接: https://juejin.cn/post/7357569755967094794