掘金后端 ( ) • 2024-04-15 13:21

本文为稀土掘金技术社区首发签约文章，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。