bingyi/gf
1
0
Fork 0
mirror of https://gitee.com/johng/gf synced 2026-07-08 14:39:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

complete pckage gtimer development, a levelled timing wheel for interval/delayed jobs running and management

Browse Source
This commit is contained in:
John
2019-01-12 22:41:12 +08:00
parent 241706cbbf
commit 432c16c89f
24 changed files with 249 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
g/container/gpool/gpool.go
View File

@ -12,7 +12,7 @@ import (
"gitee.com/johng/gf/g/container/glist"
"gitee.com/johng/gf/g/container/gtype"
"gitee.com/johng/gf/g/os/gtime"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"time"
)
@ -44,7 +44,7 @@ func New(expire int, newFunc...func() (interface{}, error)) *Pool {
if len(newFunc) > 0 {
r.NewFunc = newFunc[0]
}
gwheel.AddSingleton(time.Second, r.checkExpire)
gtimer.AddSingleton(time.Second, r.checkExpire)
return r
}
@ -102,7 +102,7 @@ func (p *Pool) Close() {
// 超时检测循环
func (p *Pool) checkExpire() {
if p.closed.Val() {
gwheel.Exit()
gtimer.Exit()
}
for {
if r := p.list.PopFront(); r != nil {

4
g/os/gcache/gcache_cache.go
View File

@ -7,7 +7,7 @@
package gcache
import (
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"sync/atomic"
"time"
"unsafe"
@ -24,7 +24,7 @@ func New(lruCap...int) *Cache {
c := &Cache {
memCache : newMemCache(lruCap...),
}
gwheel.AddSingleton(time.Second, c.syncEventAndClearExpired)
gtimer.AddSingleton(time.Second, c.syncEventAndClearExpired)
return c
}

4
g/os/gcache/gcache_mem_cache.go
View File

@ -11,7 +11,7 @@ import (
"gitee.com/johng/gf/g/container/gset"
"gitee.com/johng/gf/g/container/gtype"
"gitee.com/johng/gf/g/os/gtime"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"gitee.com/johng/gf/g/util/gconv"
"math"
"sync"
@ -292,7 +292,7 @@ func (c *memCache) syncEventAndClearExpired() {
oldExpireTime := int64(0)
newExpireTime := int64(0)
if c.closed.Val() {
gwheel.Exit()
gtimer.Exit()
return
}
// ========================

6
g/os/gcache/gcache_mem_cache_lru.go
View File

@ -11,7 +11,7 @@ import (
"gitee.com/johng/gf/g/container/glist"
"gitee.com/johng/gf/g/container/gmap"
"gitee.com/johng/gf/g/container/gtype"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"time"
)
@ -33,7 +33,7 @@ func newMemCacheLru(cache *memCache) *memCacheLru {
rawList : glist.New(),
closed : gtype.NewBool(),
}
gwheel.AddSingleton(time.Second, lru.SyncAndClear)
gtimer.AddSingleton(time.Second, lru.SyncAndClear)
return lru
}
@ -80,7 +80,7 @@ func (lru *memCacheLru) Print() {
// 异步执行协程，将queue中的数据同步到list中
func (lru *memCacheLru) SyncAndClear() {
if lru.closed.Val() {
gwheel.Exit()
gtimer.Exit()
return
}
// 数据同步

8
g/os/gcron/gcron_cron.go
View File

@ -12,7 +12,7 @@ import (
"gitee.com/johng/gf/g/container/garray"
"gitee.com/johng/gf/g/container/gmap"
"gitee.com/johng/gf/g/container/gtype"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"strconv"
"time"
)
@ -77,7 +77,7 @@ func (c *Cron) AddOnce(pattern string, job func(), name ... string) (*Entry, err
// 延迟添加定时任务，delay参数单位为秒
func (c *Cron) DelayAdd(delay int, pattern string, job func(), name ... string) {
gwheel.AddOnce(time.Duration(delay)*time.Second, func() {
gtimer.AddOnce(time.Duration(delay)*time.Second, func() {
if _, err := c.Add(pattern, job, name ...); err != nil {
panic(err)
}
@ -86,7 +86,7 @@ func (c *Cron) DelayAdd(delay int, pattern string, job func(), name ... string)
// 延迟添加单例定时任务，delay参数单位为秒
func (c *Cron) DelayAddSingleton(delay int, pattern string, job func(), name ... string) {
gwheel.AddOnce(time.Duration(delay)*time.Second, func() {
gtimer.AddOnce(time.Duration(delay)*time.Second, func() {
if _, err := c.AddSingleton(pattern, job, name ...); err != nil {
panic(err)
}
@ -95,7 +95,7 @@ func (c *Cron) DelayAddSingleton(delay int, pattern string, job func(), name ...
// 延迟添加只运行一次的定时任务，delay参数单位为秒
func (c *Cron) DelayAddOnce(delay int, pattern string, job func(), name ... string) {
gwheel.AddOnce(time.Duration(delay)*time.Second, func() {
gtimer.AddOnce(time.Duration(delay)*time.Second, func() {
if _, err := c.AddOnce(pattern, job, name ...); err != nil {
panic(err)
}

6
g/os/gcron/gcron_jobloop.go
View File

@ -7,15 +7,15 @@
package gcron
import (
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"time"
)
// 延迟添加定时任务，delay参数单位为秒
func (c *Cron) startLoop() {
gwheel.Add(time.Second, func() {
gtimer.Add(time.Second, func() {
if c.status.Val() == STATUS_CLOSED {
gwheel.Exit()
gtimer.Exit()
}
if c.status.Val() == STATUS_RUNNING {
go c.checkEntries(time.Now())

0
g/os/gcron/gcron_bench_test.go → g/os/gcron/gcron_z_bench_test.go
View File

6
g/os/gmlock/gmlock_locker.go
View File

@ -8,7 +8,7 @@ package gmlock
import (
"gitee.com/johng/gf/g/container/gmap"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"time"
)
@ -79,7 +79,7 @@ func (l *Locker) doLock(key string, expire int, try bool) bool {
if ok && expire > 0 {
// 异步goroutine计时处理
wid := mu.wid.Val()
gwheel.AddOnce(time.Duration(expire)*time.Second, func() {
gtimer.AddOnce(time.Duration(expire)*time.Second, func() {
if wid == mu.wid.Val() {
mu.Unlock()
}
@ -99,7 +99,7 @@ func (l *Locker) doRLock(key string, expire int, try bool) bool {
}
if ok && expire > 0 {
rid := mu.rid.Val()
gwheel.AddOnce(time.Duration(expire)*time.Second, func() {
gtimer.AddOnce(time.Duration(expire)*time.Second, func() {
if rid == mu.rid.Val() {
mu.RUnlock()
}

26
g/os/gwheel/gwheel.go → g/os/gtimer/gtimer.go
View File

@ -4,11 +4,11 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
// Package gwheel provides Timing Wheel for interval jobs running and management/时间轮.
// Package gtimer implements Levelled Timing Wheel for interval/delayed jobs running and management/任务定时器(分层时间轮).
// 高效的时间轮任务管理模块，用于管理间隔/延迟运行任务。
// 与gcron模块的区别是，时间轮模块只管理间隔执行任务，并且更注重执行效率(纳秒级别)。
// 需要注意执行时间间隔的准确性问题: https://github.com/golang/go/issues/14410
package gwheel
package gtimer
import (
"math"
@ -22,53 +22,53 @@ const (
gPANIC_EXIT = "exit"
gDEFAULT_TIMES = math.MaxInt32
gDEFAULT_SLOT_NUMBER = 10
gDEFAULT_WHEEL_INTERVAL = 30*time.Millisecond
gDEFAULT_WHEEL_LEVEL = 5
gDEFAULT_WHEEL_INTERVAL = 50*time.Millisecond
gDEFAULT_WHEEL_LEVEL = 6
)
var (
// 默认的wheel管理对象
defaultWheels = New(gDEFAULT_SLOT_NUMBER, gDEFAULT_WHEEL_INTERVAL, gDEFAULT_WHEEL_LEVEL)
defaultTimer = New(gDEFAULT_SLOT_NUMBER, gDEFAULT_WHEEL_INTERVAL, gDEFAULT_WHEEL_LEVEL)
)
// 添加执行方法，可以给定名字，以便于后续执行删除
func Add(interval time.Duration, job JobFunc) *Entry {
return defaultWheels.Add(interval, job)
return defaultTimer.Add(interval, job)
}
// 添加单例运行循环任务
func AddSingleton(interval time.Duration, job JobFunc) *Entry {
return defaultWheels.AddSingleton(interval, job)
return defaultTimer.AddSingleton(interval, job)
}
// 添加只运行一次的循环任务
func AddOnce(interval time.Duration, job JobFunc) *Entry {
return defaultWheels.AddOnce(interval, job)
return defaultTimer.AddOnce(interval, job)
}
// 添加运行指定次数的循环任务
func AddTimes(interval time.Duration, times int, job JobFunc) *Entry {
return defaultWheels.AddTimes(interval, times, job)
return defaultTimer.AddTimes(interval, times, job)
}
// 延迟添加循环任务，delay参数单位为秒
func DelayAdd(delay time.Duration, interval time.Duration, job JobFunc) {
defaultWheels.DelayAdd(delay, interval, job)
defaultTimer.DelayAdd(delay, interval, job)
}
// 延迟添加单例循环任务，delay参数单位为秒
func DelayAddSingleton(delay time.Duration, interval time.Duration, job JobFunc) {
defaultWheels.DelayAddSingleton(delay, interval, job)
defaultTimer.DelayAddSingleton(delay, interval, job)
}
// 延迟添加只运行一次的循环任务，delay参数单位为秒
func DelayAddOnce(delay time.Duration, interval time.Duration, job JobFunc) {
defaultWheels.DelayAddOnce(delay, interval, job)
defaultTimer.DelayAddOnce(delay, interval, job)
}
// 延迟添加运行指定次数的循环任务，delay参数单位为秒
func DelayAddTimes(delay time.Duration, interval time.Duration, times int, job JobFunc) {
defaultWheels.DelayAddTimes(delay, interval, times, job)
defaultTimer.DelayAddTimes(delay, interval, times, job)
}
// 在Job方法中调用，停止当前运行的任务

25
g/os/gwheel/gwheel_entry.go → g/os/gtimer/gtimer_entry.go
View File

@ -4,7 +4,7 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
package gwheel
package gtimer
import (
"gitee.com/johng/gf/g/container/gtype"
@ -13,7 +13,6 @@ import (
// 循环任务项
type Entry struct {
id int64 // ID
job JobFunc // 注册循环任务方法
wheel *wheel // 所属时间轮
singleton *gtype.Bool // 任务是否单例运行
@ -29,7 +28,7 @@ type Entry struct {
// 任务执行方法
type JobFunc = func()
// 创建循环任务(失败返回nil)
// 创建定时任务
func (w *wheel) addEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry {
ms := interval.Nanoseconds()/1e6
num := ms/w.intervalMs
@ -41,7 +40,6 @@ func (w *wheel) addEntry(interval time.Duration, job JobFunc, singleton bool, ti
nowMs := time.Now().UnixNano()/1e6
ticks := w.ticks.Val()
entry := &Entry {
id : time.Now().UnixNano(),
wheel : w,
singleton : gtype.NewBool(singleton),
status : gtype.NewInt(STATUS_READY),
@ -58,6 +56,18 @@ func (w *wheel) addEntry(interval time.Duration, job JobFunc, singleton bool, ti
return entry
}
// 递增添加任务
func (w *wheel) reAddEntry(entry *Entry, nowTicks int64, nowMs int64) {
left := entry.interval - (nowTicks - entry.create)
if left <= 0 {
left = entry.interval
entry.create = nowTicks
entry.createMs = nowMs
entry.updateMs = nowMs
}
w.slots[(nowTicks + left) % w.number].PushBack(entry)
}
// 获取任务状态
func (entry *Entry) Status() int {
return entry.status.Val()
@ -95,9 +105,6 @@ func (entry *Entry) Run() {
// 检测当前任务是否可运行, 参数为当前时间的纳秒数, 精度更高
func (entry *Entry) check(nowTicks int64, nowMs int64) bool {
//if entry.intervalMs == 1400 {
// fmt.Println(nowTicks, entry.create, entry.interval, time.Now())
//}
// 运行检查
if diff := nowTicks - entry.create; diff > 0 && diff%entry.interval == 0 {
// 是否单例
@ -121,9 +128,8 @@ func (entry *Entry) check(nowTicks int64, nowMs int64) bool {
}
// 分层转换
if entry.wheel.level > 0 {
if diffMs := nowMs - entry.updateMs; diffMs < entry.intervalMs {
if diffMs := nowMs - entry.updateMs; diffMs < entry.intervalMs {
if leftMs := entry.intervalMs - diffMs; leftMs > entry.wheel.wheels.intervalMs {
entry.status.Set(STATUS_CLOSED)
delay := time.Duration(leftMs)*time.Millisecond
// 往底层添加
entry.wheel.wheels.addEntry(delay, entry.job, false, 1)
@ -136,6 +142,7 @@ func (entry *Entry) check(nowTicks int64, nowMs int64) bool {
entry.job,
)
}
entry.status.Set(STATUS_CLOSED)
return false
}
}

18
g/os/gwheel/gwheel_loop.go → g/os/gtimer/gtimer_loop.go
View File

@ -4,7 +4,7 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
package gwheel
package gtimer
import (
"gitee.com/johng/gf/g/container/glist"
@ -34,13 +34,14 @@ func (w *wheel) proceed() {
l := w.slots[int(n%w.number)]
if l.Len() > 0 {
go func(l *glist.List, nowTicks int64) {
entry := (*Entry)(nil)
nowMs := time.Now().UnixNano()/1e6
for i := l.Len(); i > 0; i-- {
v := l.PopFront()
if v == nil {
if v := l.PopFront(); v == nil {
break
} else {
entry = v.(*Entry)
}
entry := v.(*Entry)
if entry.Status() == STATUS_CLOSED {
continue
}
@ -65,14 +66,7 @@ func (w *wheel) proceed() {
}
// 是否继续添运行
if entry.status.Val() != STATUS_CLOSED {
left := entry.interval - (nowTicks - entry.create)
if left <= 0 {
left = entry.interval
entry.create = nowTicks
entry.createMs = nowMs
entry.updateMs = nowMs
}
w.slots[(nowTicks + left) % w.number].PushBack(entry)
w.reAddEntry(entry, nowTicks, nowMs)
}
}
}(l, n)

174
g/os/gtimer/gtimer_timer.go Normal file
View File

@ -0,0 +1,174 @@
// Copyright 2019 gf Author(https://gitee.com/johng/gf). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
package gtimer
import (
"gitee.com/johng/gf/g/container/glist"
"gitee.com/johng/gf/g/container/gtype"
"time"
)
// 定时器/分层时间轮
type Timer struct {
wheels []*wheel // 分层
length int // 层数
number int // 每一层Slot Number
intervalMs int64 // 最小时间刻度(毫秒)
}
// 创建分层时间轮
func New(slot int, interval time.Duration, level...int) *Timer {
length := gDEFAULT_WHEEL_LEVEL
if len(level) > 0 {
length = level[0]
}
t := &Timer {
wheels : make([]*wheel, length),
length : length,
number : slot,
intervalMs : interval.Nanoseconds()/1e6,
}
for i := 0; i < length; i++ {
if i > 0 {
n := time.Duration(t.wheels[i - 1].totalMs)*time.Millisecond
w := t.newWheel(i, slot, n)
t.wheels[i] = w
t.wheels[i - 1].addEntry(n, w.proceed, false, gDEFAULT_TIMES)
} else {
t.wheels[i] = t.newWheel(i, slot, interval)
}
}
t.wheels[0].start()
return t
}
// 创建自定义的循环任务管理对象
func (t *Timer) newWheel(level int, slot int, interval time.Duration) *wheel {
w := &wheel {
wheels : t,
level : level,
slots : make([]*glist.List, slot),
number : int64(slot),
closed : make(chan struct{}, 1),
ticks : gtype.NewInt64(),
totalMs : int64(slot)*interval.Nanoseconds()/1e6,
createMs : time.Now().UnixNano()/1e6,
intervalMs : interval.Nanoseconds()/1e6,
}
for i := int64(0); i < w.number; i++ {
w.slots[i] = glist.New()
}
return w
}
// 添加循环任务
func (t *Timer) Add(interval time.Duration, job JobFunc) *Entry {
return t.addEntry(interval, job, false, gDEFAULT_TIMES)
}
// 添加单例运行循环任务
func (t *Timer) AddSingleton(interval time.Duration, job JobFunc) *Entry {
return t.addEntry(interval, job, true, gDEFAULT_TIMES)
}
// 添加只运行一次的循环任务
func (t *Timer) AddOnce(interval time.Duration, job JobFunc) *Entry {
return t.addEntry(interval, job, false, 1)
}
// 添加运行指定次数的循环任务
func (t *Timer) AddTimes(interval time.Duration, times int, job JobFunc) *Entry {
return t.addEntry(interval, job, false, times)
}
// 延迟添加循环任务
func (t *Timer) DelayAdd(delay time.Duration, interval time.Duration, job JobFunc) {
t.AddOnce(delay, func() {
t.Add(interval, job)
})
}
// 延迟添加单例循环任务
func (t *Timer) DelayAddSingleton(delay time.Duration, interval time.Duration, job JobFunc) {
t.AddOnce(delay, func() {
t.AddSingleton(interval, job)
})
}
// 延迟添加只运行一次的循环任务
func (t *Timer) DelayAddOnce(delay time.Duration, interval time.Duration, job JobFunc) {
t.AddOnce(delay, func() {
t.AddOnce(interval, job)
})
}
// 延迟添加只运行一次的循环任务
func (t *Timer) DelayAddTimes(delay time.Duration, interval time.Duration, times int, job JobFunc) {
t.AddOnce(delay, func() {
t.AddTimes(interval, times, job)
})
}
// 关闭分层时间轮
func (t *Timer) Close() {
for _, w := range t.wheels {
w.Close()
}
}
// 添加定时任务
func (t *Timer) addEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry {
intervalMs := interval.Nanoseconds()/1e6
pos, cmp := t.binSearchIndex(intervalMs)
switch cmp {
// n比最后匹配值小
case -1:
i := pos
for ; i > 0; i-- {
if intervalMs >= t.wheels[i].intervalMs && intervalMs < t.wheels[i].totalMs {
return t.wheels[i].addEntry(interval, job, singleton, times)
}
}
return t.wheels[i].addEntry(interval, job, singleton, times)
// n比最后匹配值大
case 1:
i := pos
for ; i < t.length - 1; i++ {
if intervalMs >= t.wheels[i].intervalMs && intervalMs < t.wheels[i].totalMs {
return t.wheels[i].addEntry(interval, job, singleton, times)
}
}
return t.wheels[i].addEntry(interval, job, singleton, times)
case 0:
return t.wheels[pos].addEntry(interval, job, singleton, times)
}
return nil
}
// 二分查找当前任务可以添加的时间轮对象索引
func (t *Timer) binSearchIndex(n int64)(index int, result int) {
min := 0
max := t.length - 1
mid := 0
cmp := -2
for min <= max {
mid = int((min + max) / 2)
switch {
case t.wheels[mid].intervalMs == n : cmp = 0
case t.wheels[mid].intervalMs > n : cmp = -1
case t.wheels[mid].intervalMs < n : cmp = 1
}
switch cmp {
case -1 : max = mid - 1
case 1 : min = mid + 1
case 0 :
return mid, cmp
}
}
return mid, cmp
}

4
g/os/gwheel/gwheel_wheels_wheel.go → g/os/gtimer/gtimer_wheel.go
View File

@ -4,7 +4,7 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
package gwheel
package gtimer
import (
"gitee.com/johng/gf/g/container/glist"
@ -13,7 +13,7 @@ import (
// 单层时间轮
type wheel struct {
wheels *Wheels // 所属分层时间轮
wheels *Timer // 所属定时器
level int // 所属分层索引号
slots []*glist.List // 所有的循环任务项, 按照Slot Number进行分组
number int64 // Slot Number

6
g/os/gwheel/gwheel_z_bench_test.go → g/os/gtimer/gtimer_z_bench_test.go
View File

@ -4,10 +4,10 @@
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
package gwheel_test
package gtimer_test
import (
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"testing"
"time"
)
@ -16,7 +16,7 @@ import (
func Benchmark_Add(b *testing.B) {
for i := 0; i < b.N; i++ {
// 基准测试的时候不能设置为1秒，否则大量的任务会崩掉系统
gwheel.Add(time.Hour, func() {
gtimer.Add(time.Hour, func() {
})
}

10
g/os/gwheel/gwheel_z_unit_1_test.go → g/os/gtimer/gtimer_z_unit_1_test.go
View File

@ -6,19 +6,19 @@
// 包方法操作
package gwheel_test
package gtimer_test
import (
"gitee.com/johng/gf/g/container/garray"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"gitee.com/johng/gf/g/util/gtest"
"testing"
"time"
)
func New() *gwheel.Wheels {
return gwheel.New(10, 10*time.Millisecond)
func New() *gtimer.Timer {
return gtimer.New(10, 10*time.Millisecond)
}
func TestWheel_Add_Close(t *testing.T) {
@ -142,7 +142,7 @@ func TestWheel_ExitJob(t *testing.T) {
array := garray.New(0, 0)
wheel.Add(time.Second, func() {
array.Append(1)
gwheel.Exit()
gtimer.Exit()
})
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 1)

2
g/os/gwheel/gwheel_z_unit_2_test.go → g/os/gtimer/gtimer_z_unit_2_test.go
View File

@ -6,7 +6,7 @@
// Entry操作
package gwheel_test
package gtimer_test
import (
"gitee.com/johng/gf/g/container/garray"

2
g/os/gwheel/gwheel_z_unit_3_test.go → g/os/gtimer/gtimer_z_unit_3_test.go
View File

@ -6,7 +6,7 @@
// 指定次数运行测试
package gwheel_test
package gtimer_test
import (
"gitee.com/johng/gf/g/container/garray"

174
g/os/gwheel/gwheel_wheels.go
View File

@ -1,174 +0,0 @@
// Copyright 2019 gf Author(https://gitee.com/johng/gf). All Rights Reserved.
//
// This Source Code Form is subject to the terms of the MIT License.
// If a copy of the MIT was not distributed with this file,
// You can obtain one at https://gitee.com/johng/gf.
package gwheel
import (
"gitee.com/johng/gf/g/container/glist"
"gitee.com/johng/gf/g/container/gtype"
"time"
)
// 分层时间轮
type Wheels struct {
levels []*wheel // 分层
length int // 层数
number int // 每一层Slot Number
intervalMs int64 // 最小时间刻度(毫秒)
}
// 创建分层时间轮
func New(slot int, interval time.Duration, level...int) *Wheels {
length := gDEFAULT_WHEEL_LEVEL
if len(level) > 0 {
length = level[0]
}
ws := &Wheels {
levels : make([]*wheel, length),
length : length,
number : slot,
intervalMs : interval.Nanoseconds()/1e6,
}
for i := 0; i < length; i++ {
if i > 0 {
n := time.Duration(ws.levels[i - 1].totalMs)*time.Millisecond
w := ws.newWheel(i, slot, n)
ws.levels[i] = w
ws.levels[i - 1].addEntry(n, w.proceed, false, gDEFAULT_TIMES)
} else {
ws.levels[i] = ws.newWheel(i, slot, interval)
}
}
ws.levels[0].start()
return ws
}
// 创建自定义的循环任务管理对象
func (ws *Wheels) newWheel(level int, slot int, interval time.Duration) *wheel {
w := &wheel {
wheels : ws,
level : level,
slots : make([]*glist.List, slot),
number : int64(slot),
closed : make(chan struct{}, 1),
ticks : gtype.NewInt64(),
totalMs : int64(slot)*interval.Nanoseconds()/1e6,
createMs : time.Now().UnixNano()/1e6,
intervalMs : interval.Nanoseconds()/1e6,
}
for i := int64(0); i < w.number; i++ {
w.slots[i] = glist.New()
}
return w
}
// 添加循环任务
func (ws *Wheels) Add(interval time.Duration, job JobFunc) *Entry {
return ws.addEntry(interval, job, false, gDEFAULT_TIMES)
}
// 添加单例运行循环任务
func (ws *Wheels) AddSingleton(interval time.Duration, job JobFunc) *Entry {
return ws.addEntry(interval, job, true, gDEFAULT_TIMES)
}
// 添加只运行一次的循环任务
func (ws *Wheels) AddOnce(interval time.Duration, job JobFunc) *Entry {
return ws.addEntry(interval, job, false, 1)
}
// 添加运行指定次数的循环任务
func (ws *Wheels) AddTimes(interval time.Duration, times int, job JobFunc) *Entry {
return ws.addEntry(interval, job, false, times)
}
// 延迟添加循环任务
func (ws *Wheels) DelayAdd(delay time.Duration, interval time.Duration, job JobFunc) {
ws.AddOnce(delay, func() {
ws.Add(interval, job)
})
}
// 延迟添加单例循环任务
func (ws *Wheels) DelayAddSingleton(delay time.Duration, interval time.Duration, job JobFunc) {
ws.AddOnce(delay, func() {
ws.AddSingleton(interval, job)
})
}
// 延迟添加只运行一次的循环任务
func (ws *Wheels) DelayAddOnce(delay time.Duration, interval time.Duration, job JobFunc) {
ws.AddOnce(delay, func() {
ws.AddOnce(interval, job)
})
}
// 延迟添加只运行一次的循环任务
func (ws *Wheels) DelayAddTimes(delay time.Duration, interval time.Duration, times int, job JobFunc) {
ws.AddOnce(delay, func() {
ws.AddTimes(interval, times, job)
})
}
// 关闭分层时间轮
func (ws *Wheels) Close() {
for _, w := range ws.levels {
w.Close()
}
}
// 添加循环任务
func (ws *Wheels) addEntry(interval time.Duration, job JobFunc, singleton bool, times int) *Entry {
intervalMs := interval.Nanoseconds()/1e6
pos, cmp := ws.binSearchIndex(intervalMs)
switch cmp {
// n比最后匹配值小
case -1:
i := pos
for ; i > 0; i-- {
if intervalMs > ws.levels[i].totalMs {
return ws.levels[i].addEntry(interval, job, singleton, times)
}
}
return ws.levels[i].addEntry(interval, job, singleton, times)
// n比最后匹配值大
case 1:
i := pos
for ; i < ws.length - 1; i++ {
if intervalMs < ws.levels[i].totalMs {
return ws.levels[i].addEntry(interval, job, singleton, times)
}
}
return ws.levels[i].addEntry(interval, job, singleton, times)
case 0:
return ws.levels[pos].addEntry(interval, job, singleton, times)
}
return nil
}
func (ws *Wheels) binSearchIndex(n int64)(index int, result int) {
min := 0
max := ws.length - 1
mid := 0
cmp := -2
for min <= max {
mid = int((min + max) / 2)
switch {
case ws.levels[mid].totalMs == n : cmp = 0
case ws.levels[mid].totalMs > n : cmp = -1
case ws.levels[mid].totalMs < n : cmp = 1
}
switch cmp {
case -1 : max = mid - 1
case 1 : min = mid + 1
case 0 :
return mid, cmp
}
}
return mid, cmp
}

6
geg/os/gwheel/gwheel1.go → geg/os/gtimer/gtimer1.go
View File

@ -2,14 +2,14 @@ package main
import (
"fmt"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"time"
)
func main() {
now := time.Now()
interval := 1400*time.Millisecond
gwheel.Add(interval, func() {
interval := 1000*time.Millisecond
gtimer.Add(interval, func() {
fmt.Println(time.Now(), time.Duration(time.Now().UnixNano() - now.UnixNano()))
now = time.Now()
})

10
geg/os/gwheel/gwheel2.go → geg/os/gtimer/gtimer2.go
View File

@ -3,24 +3,24 @@ package main
import (
"fmt"
"gitee.com/johng/gf/g/container/gtype"
"gitee.com/johng/gf/g/os/gwheel"
"gitee.com/johng/gf/g/os/gtimer"
"time"
)
func main() {
v := gtype.NewInt()
w := gwheel.New(10, 10*time.Millisecond)
//w := gtimer.New(10, 10*time.Millisecond)
fmt.Println("start:", time.Now())
for i := 0; i < 1000000; i++ {
w.AddOnce(time.Second, func() {
gtimer.AddOnce(time.Second, func() {
v.Add(1)
})
}
fmt.Println("end :", time.Now())
time.Sleep(3020*time.Millisecond)
time.Sleep(1300*time.Millisecond)
fmt.Println(v.Val(), time.Now())
//gwheel.AddSingleton(time.Second, func() {
//gtimer.AddSingleton(time.Second, func() {
// fmt.Println(time.Now().String())
//})
//select { }

0
geg/os/gwheel/sleep1.go → geg/os/gtimer/sleep1.go
View File

0
geg/os/gwheel/sleep2.go → geg/os/gtimer/sleep2.go
View File

0
geg/os/gwheel/ticker1.go → geg/os/gtimer/ticker1.go
View File

0
geg/os/gwheel/ticker2.go → geg/os/gtimer/ticker2.go
View File