定时任务cron

# 启动crontab
$ service crond start

# 关闭crontab
$ service crond stop

# 重启crontab
$ service crond restart

# 重载crontab
$ service crond reload

$ crontab [-u user] [-e] [-l] [-r]

minute hour day month week command

# .---------------- minute (0 - 59)
# | .------------- hour (0 - 23)
# | | .---------- day of month (1 - 31)
# | | | .------- month (1 - 12) OR jan,feb,mar,apr ...
# | | | | .---- day of week (0 - 6) (Sunday=0 or 7) OR sun,mon,tue,wed,thu,fri,sat
# | | | | |
# * * * * * user-name command to be executed

vim /etc/crontab
加上用户名 root

*/30 * * * * root date && source /root/.bash_profile &&  /usr/local/bin/ruby /data/programs/appstore_connect/main.rb >> /data/programs/appstore_connect/login.log
55 17 * * * root /bin/bash /data/programs/appstore_connect/get_appstoreconnect_data.sh >> /data/programs/appstore_connect/get_appstoreconnect_data.log

$ go get -u -v github.com/robfig/cron/v3

func TestCronDemo(t *testing.T) {
  c := cron.New()
  // 通过AddFunc注册
  c.AddFunc("30 * * * *", func() { fmt.Println("Every hour on the half hour") })
  c.AddFunc("30 3-6,20-23 * * *", func() { fmt.Println(".. in the range 3-6am, 8-11pm") })
  c.AddFunc("CRON_TZ=Asia/Tokyo 30 04 * * *", func() { fmt.Println("Runs at 04:30 Tokyo time every day") })
  c.AddFunc("@every 5m", func() { fmt.Println("every 5m, start 5m fron now") })
  // 通过AddJob注册
  // var cJob cronJobDemo
  //  c.AddJob("@every 5s", cJob)
  // 启动
  c.Start()
  // 停止
  c.Stop()
}
​
type cronJobDemo int
​
func (c cronJobDemo) Run() {
  fmt.Println("5s func trigger")
  return
}

type Option func(*Cron)

Cron = cron.New(cron.WithParser(cron.NewParser(
		cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
	)))

func (c *Cron) AddFunc(spec string, cmd func()) (EntryID, error) {
    // 包装
	return c.AddJob(spec, FuncJob(cmd))
}

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
}

// 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 数据结构，为robfig/cron的运行实体使用的s数据结构
type Cron struct {
  entries   []*Entry          // 调度执行实体列表
   // 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都能退出
}

func (c *Cron) Start() {
	c.runningMu.Lock()
	defer c.runningMu.Unlock()
	if c.running {
		return
	}
	c.running = true
	go c.run()
}

func (c *Cron) run() {
	c.logger.Info("start")

    // 第一部分
	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 {
        // 2.1
		sort.Sort(byTime(c.entries))

        // 2.2
		var timer *time.Timer
		if len(c.entries) == 0 || c.entries[0].Next.IsZero() {
			timer = time.NewTimer(100000 * time.Hour)
		} else {
			timer = time.NewTimer(c.entries[0].Next.Sub(now))
		}

        // 2.3
		for {
			select {}
			break
		}
	}
}

c.runningMu.Lock()
	if c.running {
		c.runningMu.Unlock()
		return
	}
	c.running = true
	c.runningMu.Unlock()
	c.run()

func main() {
	//启动
	go func() {
		defer func() {
			if r := recover(); r != nil {
				logger.Log.Errorf("start panic:%v stack:%s", r, string(debug.Stack()))
			}
		}()
		start()
	}()
	
	go func() {
		//定时检查是否有规则调整
		defer func() {
			if r := recover(); r != nil {
				logger.Log.Errorf("checkRules panic:%v stack:%s", r, string(debug.Stack()))
			}
		}()
		checkRules()
	}()
	
	for {
		select {}
	}
}

func start() {
	Cron = cron.New(cron.WithParser(cron.NewParser(
		cron.SecondOptional | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor,
	)))
	
	//获取有效的规则
	ruleList, err := rule.GetRuleList("", []int{rule.StatusOpen})
	if err != nil {
		logger.Log.Errorf("get rule list err:%s", err.Error())
	} else {
		for _, rule := range ruleList {
			addTask(&rule)
		}
	}
	//开启任务
	Cron.Run()
}

func addTask(data *rule.RulesList) {
	if data == nil {
		return
	}
	task, ok := taskMap[data.RuleId]
	if data.Status != rule.StatusOpen {
		//删除或者暂停规则
		if ok {
			//如果已经在任务队列中，则移除
			Cron.Remove(task.CronId)
			delete(taskMap, data.RuleId)
		}
		return
	}
	//下面的都是开启规则的逻辑

	//如果该规则已经存在于任务中
	if ok {
		//上次开启任务后，规则没有改变
		if data.LastUpdateTime.Equal(task.RuleInfo.LastUpdateTime) {
			return
		}
		//有改变，则先删除旧的任务
		Cron.Remove(task.CronId)
		delete(taskMap, data.RuleId)
	}
	checkFrequency := &controllerRule.Frequency{}
	err := json.Unmarshal([]byte(data.CheckFrequency), checkFrequency)
	if err != nil {
		return
	}
	//根据检查频次来设置任务
	//for example "30 */3 * * * ?"
	//【second】:0-59、【minute】:0-59、【hour】:0-23、【day of month】:1-31 【month】:1-12 or JAN-DEC 【day of week】:0-6 orSUN-SAT
	specs := []string{"*", "*", "*", "*", "*", "?"}
	var nextTime time.Time

	if checkFrequency.Type == rule.FrequencyTypeDay {
		//正常
		specs[0] = strconv.Itoa(data.LastUpdateTime.Second())
		specs[1] = strconv.Itoa(data.LastUpdateTime.Minute())
		specs[2] = strconv.Itoa(data.LastUpdateTime.Hour())
		specs[3] = "*/" + strconv.Itoa(checkFrequency.Period)
	} else {
		//主要为了测试用
		specs[0] = strconv.Itoa(data.LastUpdateTime.Second())
		specs[1] = "*/" + strconv.Itoa(checkFrequency.Period)
	}
	newTask := &Task{
		RuleInfo: data,
	}
	spec := strings.Join(specs, " ")
	entryId, err := Cron.AddJob(spec, newTask)
	if err != nil {
		return
	}

	timeNow := time.Now()
	if data.LastUpdateTime.Before(timeNow) {
		subDuration := time.Now().Sub(data.LastUpdateTime)
		if checkFrequency.Type == rule.FrequencyTypeDay {
			//正常
			subMinute := subDuration.Hours() / 24
			tmp := int(math.Ceil(subMinute / float64(checkFrequency.Period)))
			nextTime = data.LastUpdateTime.Add(time.Duration(tmp*tmp*checkFrequency.Period) * 24 * time.Hour)
		} else {
			//主要为了测试
			subMinute := subDuration.Minutes()
			tmp := int(math.Ceil(subMinute / float64(checkFrequency.Period)))
			nextTime = data.LastUpdateTime.Add(time.Duration(tmp*checkFrequency.Period) * time.Minute)
		}

		////测试用
		//nextTime = timeNow.Add(2 * time.Second)

		Cron.SetNextTime(entryId, nextTime)
	}

	//添加到任务映射
	newTask.CronId = entryId
	taskMap[data.RuleId] = newTask

	return
}

type Task struct {
	RuleInfo *rule.RulesList
	CronId   cron.EntryID
}

func (t Task) Run() {
	defer func() {
		if r := recover(); r != nil {
			dingding.SendDdMessage("", fmt.Sprintf("run panic:%s", r))
		}
	}()
	logger.Log.Infof("cronId:%d ***** run start *****:%s", t.CronId, time.Now().Format("2006-01-02 15:04:05"))

	err := run(t.RuleInfo)
	if err != nil {
		dingding.SendDdMessage("", fmt.Sprintf("ruleId:%d run fail", t.RuleInfo.RuleId))
	}
	logger.Log.Infof("cronId:%d ***** run end *****:%s", t.CronId, time.Now().Format("2006-01-02 15:04:05"))
}

func (c *Cron) SetNextTime(id EntryID, nextTime time.Time) {
	c.runningMu.Lock()
	defer c.runningMu.Unlock()
	if c.running {
		return
	}
	if nextTime.Before(time.Now()) {
		return
	}
	for _, e := range c.entries {
		if e.ID != id {
			continue
		}
		e.Next = nextTime
	}
}