zinx/ztimer/timerscheduler.go

package ztimer

/**
* @Author: Aceld
* @Date: 2019/5/8 17:43
* @Mail: danbing.at@gmail.com
*
*  时间轮调度器
*   依赖模块，delayfunc.go  timer.go timewheel.go
 */

import (
	"math"
	"sync"
	"time"

	"github.com/aceld/zinx/zlog"
)

const (
	//MaxChanBuff 默认缓冲触发函数队列大小
	MaxChanBuff = 2048
	//MaxTimeDelay 默认最大误差时间
	MaxTimeDelay = 100
)

//TimerScheduler 计时器调度器
type TimerScheduler struct {
	//当前调度器的最高级时间轮
	tw *TimeWheel
	//定时器编号累加器
	IDGen uint32
	//已经触发定时器的channel
	triggerChan chan *DelayFunc
	//互斥锁
	sync.RWMutex
	//所有注册的timerID集合
	IDs []uint32
}

// NewTimerScheduler 返回一个定时器调度器 ，主要创建分层定时器，并做关联，并依次启动
func NewTimerScheduler() *TimerScheduler {

	//创建秒级时间轮
	secondTw := NewTimeWheel(SecondName, SecondInterval, SecondScales, TimersMaxCap)
	//创建分钟级时间轮
	minuteTw := NewTimeWheel(MinuteName, MinuteInterval, MinuteScales, TimersMaxCap)
	//创建小时级时间轮
	hourTw := NewTimeWheel(HourName, HourInterval, HourScales, TimersMaxCap)

	//将分层时间轮做关联
	hourTw.AddTimeWheel(minuteTw)
	minuteTw.AddTimeWheel(secondTw)

	//时间轮运行
	secondTw.Run()
	minuteTw.Run()
	hourTw.Run()

	return &TimerScheduler{
		tw:          hourTw,
		triggerChan: make(chan *DelayFunc, MaxChanBuff),
		IDs:         make([]uint32, 0),
	}
}

//CreateTimerAt 创建一个定点Timer 并将Timer添加到分层时间轮中， 返回Timer的tID
func (ts *TimerScheduler) CreateTimerAt(df *DelayFunc, unixNano int64) (uint32, error) {
	ts.Lock()
	defer ts.Unlock()

	ts.IDGen++
	ts.IDs = append(ts.IDs, ts.IDGen)
	return ts.IDGen, ts.tw.AddTimer(ts.IDGen, NewTimerAt(df, unixNano))
}

//CreateTimerAfter 创建一个延迟Timer 并将Timer添加到分层时间轮中， 返回Timer的tID
func (ts *TimerScheduler) CreateTimerAfter(df *DelayFunc, duration time.Duration) (uint32, error) {
	ts.Lock()
	defer ts.Unlock()

	ts.IDGen++
	ts.IDs = append(ts.IDs, ts.IDGen)
	return ts.IDGen, ts.tw.AddTimer(ts.IDGen, NewTimerAfter(df, duration))
}

//CancelTimer 删除timer
func (ts *TimerScheduler) CancelTimer(tID uint32) {
	ts.Lock()
	ts.Unlock()
	//ts.tw.RemoveTimer(tID)  这个方法无效
	//删除timerID
	var index = -1
	for i := 0; i < len(ts.IDs); i++ {
		if ts.IDs[i] == tID {
			index = i
		}
	}

	if index > -1 {
		ts.IDs = append(ts.IDs[:index], ts.IDs[index+1:]...)
	}
}

//GetTriggerChan 获取计时结束的延迟执行函数通道
func (ts *TimerScheduler) GetTriggerChan() chan *DelayFunc {
	return ts.triggerChan
}

// HasTimer 是否有时间轮
func (ts *TimerScheduler) HasTimer(tID uint32) bool {
	for i := 0; i < len(ts.IDs); i++ {
		if ts.IDs[i] == tID {
			return true
		}
	}
	return false
}

//Start 非阻塞的方式启动timerSchedule
func (ts *TimerScheduler) Start() {
	go func() {
		for {
			//当前时间
			now := UnixMilli()
			//获取最近MaxTimeDelay 毫秒的超时定时器集合
			timerList := ts.tw.GetTimerWithIn(MaxTimeDelay * time.Millisecond)
			for tID, timer := range timerList {
				if math.Abs(float64(now-timer.unixts)) > MaxTimeDelay {
					//已经超时的定时器，报警
					zlog.Error("want call at ", timer.unixts, "; real call at", now, "; delay ", now-timer.unixts)
				}
				if ts.HasTimer(tID) {
					//将超时触发函数写入管道
					ts.triggerChan <- timer.delayFunc
				}
			}
			time.Sleep(MaxTimeDelay / 2 * time.Millisecond)
		}
	}()
}

//NewAutoExecTimerScheduler 时间轮定时器 自动调度
func NewAutoExecTimerScheduler() *TimerScheduler {
	//创建一个调度器
	autoExecScheduler := NewTimerScheduler()
	//启动调度器
	autoExecScheduler.Start()

	//永久从调度器中获取超时 触发的函数 并执行
	go func() {
		delayFuncChan := autoExecScheduler.GetTriggerChan()
		for df := range delayFuncChan {
			go df.Call()
		}
	}()

	return autoExecScheduler
}