通过源码分析Golang cron的实现原理

来到golang学习网的大家，相信都是编程学习爱好者，希望在这里学习Golang相关编程知识。下面本篇文章就来带大家聊聊《通过源码分析Golang cron的实现原理》，介绍一下cron，希望对大家的知识积累有所帮助，助力实战开发！

关于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 
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 = 
Stop()实现
Stop() 用来停止Cron的运行，但已经在执行中的作业是不会被打断的，也就是从执行 Stop() 之后，不会再有新的任务被调度：
func (c *Cron) Stop() context.Context {
	c.runningMu.Lock()
	defer c.runningMu.Unlock()
	if c.running {
		c.stop 
Remove()实现
Remove() 用于移除一个任务：
func (c *Cron) Remove(id EntryID) {
	c.runningMu.Lock()
	defer c.runningMu.Unlock()
	if c.running {
		c.remove 
小结

在Go Cron内部维护了两个结构体Cron和Entry，用于维护任务数据，cron.Start()执行后，cron的后台程序c.Run()就开始执行了，Run()是整个cron的一个核心，它负责处理cron开始执行后的大部分事情， run中会一直轮循c.entries中的entry, 每个entry都包含自己下一次执行的绝对时间，如果一个entry 允许执行了，就会开启单独的goroutine去执行这个任务。
文中关于golang的知识介绍，希望对你的学习有所帮助！若是受益匪浅，那就动动鼠标收藏这篇《通过源码分析Golang cron的实现原理》文章吧，也可关注golang学习网公众号了解相关技术文章。