likes
comments
collection
share

golang定时任务之cron

作者站长头像
站长
· 阅读数 60

简介

github.com/robfig/cron

服务器应用中，定时任务是常见的业务需求。一些清理动作、统计、重试等耗时长、占用资源多的应用，期望在业务低峰期定时执行。通常根据需求，分为每天/每周/每月执行。cron是golang中广泛使用的一个开源项目

demo try

package main

import (
	"fmt"
	"time"

	"github.com/robfig/cron/v3"
)

func main() {
	c := cron.New()
	c.AddFunc("@every 5s", func() {
		fmt.Println(time.Now(), "run every 5 second")
	})
	c.AddFunc("4 * * * ?", func() {
		fmt.Println(time.Now(), "run minute 4 every hour")
	})
	c.Start()
	select {}
}

输出：

2024-01-23 22:03:30.001179 +0800 CST m=+4.163984251 run every 5 second
2024-01-23 22:03:35.000493 +0800 CST m=+9.163356209 run every 5 second
2024-01-23 22:03:40.001159 +0800 CST m=+14.164079876 run every 5 second
2024-01-23 22:03:45.001147 +0800 CST m=+19.164125626 run every 5 second
2024-01-23 22:03:50.001245 +0800 CST m=+24.164280209 run every 5 second
2024-01-23 22:03:55.001376 +0800 CST m=+29.164468959 run every 5 second
2024-01-23 22:04:00.004014 +0800 CST m=+34.167164418 run minute 4 every hour
2024-01-23 22:04:00.004155 +0800 CST m=+34.167305751 run every 5 second
2024-01-23 22:04:05.001184 +0800 CST m=+39.164391918 run every 5 second

一探究竟

类图

golang定时任务之cron

cron

Cron是主类，其中主要属性：
  • entries记录了任务列表，包含了任务的id、信息等
  • stop、add、remove，通过chan方式，传递job信息的添加、删除操作和cron的stop
  • Chain，装饰者模式，为cron的job提供多个jobWrapper，可以实现log、同步等功能
  • running记录当前cron的运行状态
  • runningMu，运行锁，在添加、删除、启动、停止时，均需获取runningMu锁
  • parser为cron表达式的格式，默认以“Minute | Hour | Dom | Month | Dow | Descriptor“的格式，详情可参考en.wikipedia.org/wiki/Cron
  • nextId记录当前最大的jobId
    Cron提供的主要方法：
  • 添加任务：可以通过AddFunc、AddJob直接添加待执行任务，或者通过自定义spec parse初始化schedule，通过Schedule func添加任务
  • 删除任务：Remove
  • Start: 启动全部定时任务，cron的启动需要显示启动
  • Stop: 停止cron。会等待所有正在执行的任务完毕stop
    定时任务的调度机制

    当调用Cron.Start()方法后，会调用Cron.run()

  • 首先计算每个任务的下一个运行时间点 
    	// 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)
	}
  • 然后进入循环任务，在循环任务中，首先根据任务的下次执行时间，进行升序排列 
    sort.Sort(byTime(c.entries))
  • 设置定时器，下一个任务的执行时间开始提醒
  • 检测增、删、停等channel是否有信号，进行任务管理操作 
    	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)
  • 当到达下个任务的执行时间，执行该任务，设置该任务的下次执行时间 
    	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)
		}
  • 执行任务时，新建goroutine的方式来执行该任务，故若同一个任务，不同批次的执行不会相互影响 
    	c.jobWaiter.Add(1)
	go func() {
		defer c.jobWaiter.Done()
		j.Run()
	}()

    配置项

    cron提供了一些常用配置项，供开发者自行设置

  • 时区信息: WithLocation(loc *time.Location)
  • 支持秒级任务定时: WithSeconds()
  • 支持自定义cron表达式格式: WithParser(p Parse) Option
  • 支持自定义Job的封装方法，可以自定义日志、同步机制、甚至引入prometheus监控等: WithChain(wrappers ...JobWrapper)
  • 支持自定义Logger: WithLogger(logger Logger)

    总结

    整体来说，cron通过简单清晰的任务调度机制，实现了定时任务的执行。同时，也提供了丰富的定制化功能。使用者可以根据自身需要进行扩展。

    • 转载自:https://juejin.cn/post/7330904102023036955
    评论
    请登录