Golang分布式应用定时任务示例详解

// 运行
func (c *Cron) Run() error {
    // 设置cron已启动，atomic.Bool来保证并发安全
	c.started.Store(true)
    // 主循环
	for {
        // 如果停止则退出
		if !c.started.Load() {
			break
		}
		c.runTask()
	}
	return nil
}
// 核心逻辑
func (c *Cron) runTask() {
	now := time.Now()
	duration := infTime
	// 获取堆顶元素
	task, ok := c.tasks.Peek()
	if ok {
		// 如果已删除则弹出
		if !c.set.Has(task.Name()) {
			c.tasks.Pop()
			return
		}
		// 计算于当前时间查找，设置定时器
		if task.next.After(now) {
			duration = task.next.Sub(now)
		} else {
			duration = 0
		}
	}
	timer := time.NewTimer(duration)
	defer timer.Stop()
	// 当有新元素插入直接返回，防止新元素执行时间小于当前堆顶元素
	select {
	case <-c.new:
		return
	case <-timer.C:
	}
	// 弹出任务，执行
	go task.Exec()
	// 计算下次执行时间，如果为0说明任务已结束，否则重新入堆
	task.next = task.Next(time.Now())
	if task.next.IsZero() {
		c.set.Delete(task.Name())
	} else {
		c.tasks.Push(task)
	}
}

// 定义
type TimeWheel struct {
	interval    time.Duration // 触发间隔
	slots       int // 总插槽数
	currentSlot int // 当前插槽数
	tasks       []*list.List // 环形列表，每个元素为对应插槽的任务列表
	set         containerx.Set[string] // 记录所有任务key值，用来检查任务是否被删除
	tricker *time.Ticker // 定时触发器
	logger logr.Logger
}
func (tw *TimeWheel) Run() error {
	tw.tricker = time.NewTicker(tw.interval)
	for {
		// 通过定时器模拟时间轮转动
		now, ok := <-tw.tricker.C
		if !ok {
			break
		}
		// 转动一次，执行任务列表
		tw.RunTask(now, tw.currentSlot)
		tw.currentSlot = (tw.currentSlot + 1) % tw.slots
	}
	return nil
}
func (tw *TimeWheel) RunTask(now time.Time, slot int) {
	// 一次执行任务列表
	for item := taskList.Front(); item != nil; {
		task, ok := item.Value.(*TimeWheelTask)
		// 任务圈数大于0，不需要执行，将圈数减一
		if task.circle > 0 {
			task.circle--
			item = item.Next()
			continue
		}
		// 运行任务
		go task.Exec()
		// 计算任务下次运行时间
		next := item.Next()
		taskList.Remove(item)
		item = next
		task.next = task.Next(now)
		if !task.next.IsZero() {
			tw.add(now, task)
		} else {
			tw.Remove(task.Name())
		}
	}
}
// 添加任务，计算下一次任务执行的插槽与圈数
func (tw *TimeWheel) add(now time.Time, task *TimeWheelTask) {
	if !task.initialized {
		task.next = task.Next(now)
		task.initialized = true
	}
	duration := task.next.Sub(now)
	if duration <= 0 {
		task.slot = tw.currentSlot + 1
		task.circle = 0
	} else {
		mult := int(duration / tw.interval)
		task.slot = (tw.currentSlot + mult) % tw.slots
		task.circle = mult / tw.slots
	}
	tw.tasks[task.slot].PushBack(task)
	tw.set.Insert(task.Name())
}