开发者

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

目录
  • 前言
  • Demo示例
  • 源码实现
    • 结构体 Cron 和js Entry
    • New()实现
    • AddFunc()实现
    • Start()实现
    • Run()实现
    • Stop()实现
    • Remove()实现
  • 小结

    前言

    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 之后不会再有未执行的作业被唤醒,但已经开始执行的作业不会受影响。

    源码实现

    在了解其整体逻辑的实现过程前,先了解两个重要的结构体EntryCron

    位置在/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 ...Opti编程客栈on) *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 mHchHaCGRV{
    		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 结构体,并把它追加到 Cronentries 列表中,如果 Cron 已经处于运行状态,会将这个创建好的 entry 发送到 Cronadd chan 中,在run()中会处理这种情况。

    Start()实现

    Start() 用于开始执行 Cron,其代码实现如下:

    func (c *Cron) Start()www.devze.com {
    	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.runn开发者_Python入门ing 置为 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
    }
    

    小结

    到此这篇关于Golang Cron 定时任务的内部实现的文章就介绍到这了, 其中重点如下:

    Go Cron内部维护了两个结构体CronEntry,用于维护任务数据,cron.Start()执行后,cron的后台程序c.Run()就开始执行了,Run()是整个cron的一个核心,它负责处理cron开始执行后的大部分事情, run中会一直轮循c.entries中的entry, 每个entry都包含自己下一次执行的绝对时间,如果一个entry 允许执行了,就会开启单独的goroutine去执行这个任务。

    以上就是通过源码分析Golang cron的实现原理的详细内容,更多关于Golang cron的资料请关注我们其它相关文章!

    0

    上一篇:

    下一篇:

    精彩评论

    暂无评论...
    验证码 换一张
    取 消

    最新开发

    开发排行榜