init commit

utils/timer/constants.go

@@ -0,0 +1,8 @@
+package timer
+
+import "time"
+
+var (
+	// 时间轮一刻度长度
+	timingWheelTick = time.Millisecond * 10
+)

utils/timer/manager.go

@@ -0,0 +1,104 @@
+package timer
+
+import (
+	"reflect"
+	"sync"
+	"time"
+
+	"github.com/RussellLuo/timingwheel"
+)
+
+// Manager 管理器
+type Manager struct {
+	timer  *Timer
+	wheel  *timingwheel.TimingWheel
+	timers map[string]*Scheduler
+	lock   sync.RWMutex
+}
+
+// Release 释放管理器，并将管理器重新放回 Timer 池中
+func (slf *Manager) Release() {
+	slf.timer.lock.Lock()
+	defer slf.timer.lock.Unlock()
+
+	slf.lock.Lock()
+
+	for name, scheduler := range slf.timers {
+		scheduler.close()
+		delete(slf.timers, name)
+	}
+
+	slf.lock.Unlock()
+
+	slf.timer.Managers = append(slf.timer.Managers, slf)
+}
+
+// StopTimer 停止特定名称的调度器
+func (slf *Manager) StopTimer(name string) {
+	slf.lock.Lock()
+	defer slf.lock.Unlock()
+
+	if s, ok := slf.timers[name]; ok {
+		s.close()
+		delete(slf.timers, name)
+	}
+}
+
+// IsStopped 特定名称的调度器是否已停止
+func (slf *Manager) IsStopped(name string) bool {
+	slf.lock.RLock()
+	defer slf.lock.RUnlock()
+	if s, ok := slf.timers[name]; ok {
+		return s.isClosed()
+	}
+	return true
+}
+
+// GetSchedulers 获取所有调度器名称
+func (slf *Manager) GetSchedulers() []string {
+	slf.lock.RLock()
+	defer slf.lock.RUnlock()
+	names := make([]string, 0, len(slf.timers))
+	for name := range slf.timers {
+		names = append(names, name)
+	}
+	return names
+}
+
+// After 设置一个在特定时间后运行一次的调度器
+func (slf *Manager) After(name string, after time.Duration, handleFunc interface{}, args ...interface{}) {
+	slf.Loop(name, after, timingWheelTick, 1, handleFunc, args...)
+}
+
+// Loop 设置一个在特定时间后仿佛运行的调度器
+func (slf *Manager) Loop(name string, after, interval time.Duration, times int, handleFunc interface{}, args ...interface{}) {
+	slf.StopTimer(name)
+
+	if after < timingWheelTick {
+		after = timingWheelTick
+	}
+	if interval < timingWheelTick {
+		interval = timingWheelTick
+	}
+
+	var values = make([]reflect.Value, len(args))
+	for i, v := range args {
+		values[i] = reflect.ValueOf(v)
+	}
+
+	scheduler := &Scheduler{
+		name:     name,
+		after:    after,
+		interval: interval,
+		total:    times,
+		cbFunc:   reflect.ValueOf(handleFunc),
+		cbArgs:   values,
+		manager:  slf,
+	}
+
+	slf.lock.Lock()
+	slf.timers[name] = scheduler
+	scheduler.timer = slf.wheel.ScheduleFunc(scheduler, scheduler.caller)
+	slf.lock.Unlock()
+}
+

utils/timer/scheduler.go

@@ -0,0 +1,95 @@
+package timer
+
+import (
+	"reflect"
+	"sync"
+	"time"
+
+	"github.com/RussellLuo/timingwheel"
+)
+
+// Scheduler 调度器
+type Scheduler struct {
+	name     string
+	after    time.Duration
+	interval time.Duration
+
+	total   int
+	trigger int
+	kill    bool
+
+	cbFunc reflect.Value
+	cbArgs []reflect.Value
+
+	timer *timingwheel.Timer
+
+	manager *Manager
+
+	lock sync.RWMutex
+}
+
+// Name 获取调度器名称
+func (slf *Scheduler) Name() string {
+	return slf.name
+}
+
+// Next 获取下一次执行的时间
+func (slf *Scheduler) Next(prev time.Time) time.Time {
+	slf.lock.RLock()
+	defer slf.lock.RUnlock()
+
+	if slf.kill || (slf.total > 0 && slf.trigger >= slf.total) {
+		return time.Time{}
+	}
+	if slf.trigger == 0 {
+		return prev.Add(slf.after)
+	}
+	return prev.Add(slf.interval)
+}
+
+// 实际执行的任务
+func (slf *Scheduler) caller() {
+	// TODO: 直接调用可能会导致更高的并发复杂度
+	slf.Caller()
+}
+
+// Caller 可由外部发起调用的执行函数
+func (slf *Scheduler) Caller() {
+	slf.lock.Lock()
+
+	if slf.kill {
+		slf.lock.Unlock()
+		return
+	}
+
+	slf.trigger++
+	if slf.total > 0 && slf.trigger >= slf.total {
+		slf.lock.Unlock()
+		slf.manager.StopTimer(slf.name)
+	} else {
+		slf.lock.Unlock()
+	}
+	slf.cbFunc.Call(slf.cbArgs)
+}
+
+// isClosed 检查调度器是否已关闭
+func (slf *Scheduler) isClosed() bool {
+	slf.lock.RLock()
+	defer slf.lock.RUnlock()
+
+	return slf.kill
+}
+
+// close 关闭调度器
+func (slf *Scheduler) close() {
+	slf.lock.Lock()
+	defer slf.lock.Unlock()
+
+	if slf.kill {
+		return
+	}
+	slf.kill = true
+	if slf.total <= 0 || slf.trigger < slf.total {
+		slf.timer.Stop()
+	}
+}

utils/timer/timer.go

@@ -0,0 +1,38 @@
+package timer
+
+import (
+	"sync"
+
+	"github.com/RussellLuo/timingwheel"
+)
+
+var timer = new(Timer)
+
+func GetManager(size int) *Manager {
+	return timer.NewManager(size)
+}
+
+type Timer struct {
+	Managers []*Manager
+	lock     sync.Mutex
+}
+
+func (slf *Timer) NewManager(size int) *Manager {
+	slf.lock.Lock()
+	defer slf.lock.Unlock()
+
+	var manager *Manager
+	if len(slf.Managers) > 0 {
+		manager = slf.Managers[0]
+		slf.Managers = slf.Managers[1:]
+		return manager
+	}
+
+	manager = &Manager{
+		timer:  slf,
+		wheel:  timingwheel.NewTimingWheel(timingWheelTick, int64(size)),
+		timers: make(map[string]*Scheduler),
+	}
+	manager.wheel.Start()
+	return manager
+}