通过源码分析Golang cron的实现原理
作者:1个俗人 发布时间:2023-06-15 23:49:24
前言
golang实现定时任务很简单,只须要简单几步代码即可以完成,最近在做了几个定时任务,想研究一下它内部是怎么实现的,所以将源码过了一遍,记录和分享在此。需要的朋友可以参考以下内容,希望对大家有帮助。
关于go cron是如何使用的可以参考之前的文章:一文带你入门Go语言中定时任务库Cron的使用
Demo示例
package main
import (
"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
里保存了这个任务上次运行的时间和下次运行的时间。具体代码实现如下:
// 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
对象指针,其实现如下:
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()
方法;其代码实现如下:
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
,其代码实现如下:
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
去执行这个任务。
// 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()
之后,不会再有新的任务被调度:
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() 用于移除一个任务:
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
}
小结
在Go Cron
内部维护了两个结构体Cron
和Entry
,用于维护任务数据,cron.Start()
执行后,cron
的后台程序c.Run()
就开始执行了,Run()
是整个cron
的一个核心,它负责处理cron
开始执行后的大部分事情, run
中会一直轮循c.entries
中的entry
, 每个entry都包含自己下一次执行的绝对时间,如果一个entry
允许执行了,就会开启单独的goroutine
去执行这个任务。
来源:https://juejin.cn/post/7155769581255000101
猜你喜欢
- 这篇论坛文章(赛迪网技术社区)主要介绍了一种简单的MySQL数据库安装方法,详细内容请大家参考下文:虽然安装MySQL数据库的文章很多,但是
- 大家一定使用过 phpmyadmin 里面的数据库导入,导出功能,非常方便。但是在实际应用中,我发现如下几个问题: 1、数据库超过一定尺寸,
- 如果我们的web应用有大量的异步请求,而这些异步请求是在web服务器认证的情况下,那当我们请求发生在服务器认证失效下,服务器自动302到登录
- 问:怎样才能指定MySQL只监听某个特定地址?答:比较常见的办法是,在my.cnf之mysqld节,添加bind-address=127.0
- 在上一个文章里写了关于左(右)侧定宽右(左)侧自动缩放的两列浮动,这个文章就要说一下三列浮动的问题了。在之前说过,两列浮动是其他多列浮动的基
- 不是很难哦,我们现在就可以实现VBScript在服务器端验证,就是下面的函数:<%FUNCTION TestString(S
- 单例模式Singleton Pattern是什么单例模式是一种创建型模式,它保证一个类只有一个实例,并提供一个访问它的全局访问点。单例模式的
- 一、数组的创建方式一var a = new Array(); a[0]="wo"
- 看了很多介绍javascript面向对象技术的文章,很晕.为什么?不是因为写得不好,而是因为太深奥.javascript中的对象还没解释清楚
- 事件流浏览器中的事件流意味着页面上可有不仅一个,甚至多个元素响应同一个事件。而这一个或多个元素响应事件发生的先后顺序在各个浏览器(主要针对I
- Howdy, 大家好,又是我~ 上一次我们简单的谈了一下font set和一些要注意的基本问题。今天我们继续字体这一话题,深入讲讲上次提到
- 如何在网站上提供音乐下载?为用户提供歌曲下载,一般有两种方式,一是直接通过Http,浏览器下载,二是通过ftp协议下载。我们来用Http和浏
- 在以前的日志中讲了怎么制作验证码,这篇就讲讲怎么给验证码加上起干扰效果的杂点。 其实很简单,首先做一个
- DEMO:var testobj = document.getElementById("test&q
- 最近项目很忙没机会更新博客。有朋友在和我谈学艺术的就业问题,就随便谈一下自己的想法。每年中国美院报考的人数与日俱增,越来越多的人投入到艺术设
- 关于建立索引的几个准则:1、合理的建立索引能够加速数据读取效率,不合理的建立索引反而会拖慢数据库的响应速度。2、索引越多,更新数据的速度越慢
- Microsoft SQL server2000由一系列相互协作的组件构成。能满足最大的WEB站点和企业数据处理系统存储
- 确定数据库内有多少记录,或者确定有多少记录达到了某些标准,这些用ASP完成并非难事。如果你采用了正确的游标类型,你可以用RecordCoun
- 如何用Access加密页面?很简单哦,看看这个用用Access和ASP做的加密程序: <%userid =&nbs
- 1、之前那个说淘宝交易评价“有问题”的文章已经有了一百多个评论。(某些开口就骂的评论已被删除)近期去评论的人都是从搜索引擎过来的,他们不是产