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

gcron updates

Browse Source
This commit is contained in:
John
2019-01-16 22:34:22 +08:00
parent 72c7e65dfa
commit 36199334f0
6 changed files with 231 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
g/os/gcron/gcron.go
View File

@ -9,15 +9,15 @@
// 定时任务.
package gcron
const (
MODE_NORMAL = 0
MODE_SINGLETON = 1
MODE_ONCE = 2
import "math"
const (
STATUS_READY = 0
STATUS_RUNNING = 1
STATUS_STOPPED = 2
STATUS_CLOSED = -1
gDEFAULT_TIMES = math.MaxInt32
)
var (
@ -40,6 +40,11 @@ func AddOnce(pattern string, job func(), name ... string) (*Entry, error) {
return defaultCron.AddOnce(pattern, job, name...)
}
// 添加运行指定次数的定时任务
func AddTimes(pattern string, times int, job func(), name ... string) (*Entry, error) {
return defaultCron.AddTimes(pattern, times, job, name...)
}
// 延迟添加定时任务，delay参数单位为秒
func DelayAdd(delay int, pattern string, job func(), name ... string) {
defaultCron.DelayAdd(delay, pattern, job, name...)
@ -55,6 +60,11 @@ func DelayAddOnce(delay int, pattern string, job func(), name ... string) {
defaultCron.DelayAddOnce(delay, pattern, job, name...)
}
// 延迟添加运行指定次数的定时任务，delay参数单位为秒
func DelayAddTimes(delay int, pattern string, times int, job func(), name ... string) {
defaultCron.DelayAddTimes(delay, pattern, times, job, name...)
}
// 检索指定名称的定时任务
func Search(name string) *Entry {
return defaultCron.Search(name)

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

@ -49,7 +49,7 @@ func (c *Cron) Add(pattern string, job func(), name ... string) (*Entry, error)
return nil, errors.New(fmt.Sprintf(`cron job "%s" already exists`, name[0]))
}
}
entry, err := newEntry(pattern, job, name ...)
entry, err := newEntry(pattern, job, false, gDEFAULT_TIMES, name ...)
if err != nil {
return nil, err
}
@ -67,7 +67,7 @@ func (c *Cron) AddSingleton(pattern string, job func(), name ... string) (*Entry
if entry, err := c.Add(pattern, job, name ...); err != nil {
return nil, err
} else {
entry.SetMode(MODE_SINGLETON)
entry.SetSingleton(true)
return entry, nil
}
}
@ -77,7 +77,17 @@ func (c *Cron) AddOnce(pattern string, job func(), name ... string) (*Entry, err
if entry, err := c.Add(pattern, job, name ...); err != nil {
return nil, err
} else {
entry.SetMode(MODE_ONCE)
entry.SetTimes(1)
return entry, nil
}
}
// 添加运行指定次数的定时任务
func (c *Cron) AddTimes(pattern string, times int, job func(), name ... string) (*Entry, error) {
if entry, err := c.Add(pattern, job, name ...); err != nil {
return nil, err
} else {
entry.SetTimes(times)
return entry, nil
}
}
@ -100,7 +110,7 @@ func (c *Cron) DelayAddSingleton(delay int, pattern string, job func(), name ...
})
}
// 延迟添加只运行一次的定时任务，delay参数单位为秒
// 延迟添加运行指定次数的定时任务，delay参数单位为秒
func (c *Cron) DelayAddOnce(delay int, pattern string, job func(), name ... string) {
gtimer.AddOnce(time.Duration(delay)*time.Second, func() {
if _, err := c.AddOnce(pattern, job, name ...); err != nil {
@ -109,6 +119,15 @@ func (c *Cron) DelayAddOnce(delay int, pattern string, job func(), name ... stri
})
}
// 延迟添加只运行一次的定时任务，delay参数单位为秒
func (c *Cron) DelayAddTimes(delay int, pattern string, times int, job func(), name ... string) {
gtimer.AddOnce(time.Duration(delay)*time.Second, func() {
if _, err := c.AddTimes(pattern, times, job, name ...); err != nil {
panic(err)
}
})
}
// 检索指定名称的定时任务
func (c *Cron) Search(name string) *Entry {
if v := c.entries.Get(name); v != nil {
@ -184,33 +203,45 @@ func (c *Cron) Entries() []*Entry {
// 遍历检查可执行定时任务，并异步执行
func (c *Cron) checkEntries(t time.Time) {
removeKeys := make([]string, 0)
c.entries.RLockFunc(func(m map[string]interface{}) {
for _, v := range m {
for k, v := range m {
entry := v.(*Entry)
if entry.schedule.meet(t) {
// 是否已命令停止运行
// 是否停止
if entry.status.Val() == STATUS_STOPPED {
continue
}
switch entry.mode.Val() {
// 是否只允许单例运行
case MODE_SINGLETON:
if entry.status.Set(STATUS_RUNNING) == STATUS_RUNNING {
continue
}
// 只运行一次的任务
case MODE_ONCE:
if entry.status.Set(STATUS_CLOSED) == STATUS_CLOSED {
continue
}
// 是否关闭
if entry.status.Val() == STATUS_CLOSED {
removeKeys = append(removeKeys, k)
continue
}
// 单例模式
if entry.status.Set(STATUS_RUNNING) == STATUS_RUNNING {
if entry.IsSingleton() {
continue
}
}
// 运行次数
if t := entry.times.Add(-1); t <= 0 {
if entry.status.Set(STATUS_CLOSED) == STATUS_CLOSED {
continue
}
} else if t < 2000000000 && t > 1000000000 {
entry.times.Set(gDEFAULT_TIMES)
}
// 执行异步运行
go func() {
defer func() {
if entry.status.Val() != STATUS_CLOSED {
entry.status.Set(STATUS_READY)
} else {
c.Remove(entry.Name)
switch entry.status.Val() {
case STATUS_CLOSED:
c.Remove(entry.Name)
case STATUS_STOPPED:
default:
entry.status.Set(STATUS_READY)
}
}()
entry.Job()
@ -218,4 +249,7 @@ func (c *Cron) checkEntries(t time.Time) {
}
}
})
if len(removeKeys) > 0 {
c.entries.BatchRemove(removeKeys)
}
}

46
g/os/gcron/gcron_entry.go
View File

@ -15,23 +15,25 @@ import (
// 定时任务项
type Entry struct {
mode *gtype.Int // 任务运行模式(0: normal; 1: singleton; 2: once)
status *gtype.Int // 定时任务状态(0: ready; 1: running; -1: stopped)
schedule *cronSchedule // 定时任务配置对象
Name string // 定时任务名称
Job func() // 注册定时任务方法
JobName string // 注册定时任务名称
Time time.Time // 注册时间
singleton *gtype.Bool // 任务是否单例运行
times *gtype.Int // 还需运行次数
status *gtype.Int // 定时任务状态(0: ready; 1: running; -1: stopped)
schedule *cronSchedule // 定时任务配置对象
Name string // 定时任务名称
Job func() // 注册定时任务方法
JobName string // 注册定时任务名称
Time time.Time // 注册时间
}
// 创建定时任务
func newEntry(pattern string, job func(), name ... string) (*Entry, error) {
func newEntry(pattern string, job func(), singleton bool, times int, name ... string) (*Entry, error) {
schedule, err := newSchedule(pattern)
if err != nil {
return nil, err
}
entry := &Entry {
mode : gtype.NewInt(),
singleton : gtype.NewBool(singleton),
times : gtype.NewInt(times),
status : gtype.NewInt(),
schedule : schedule,
Job : job,
@ -44,9 +46,19 @@ func newEntry(pattern string, job func(), name ... string) (*Entry, error) {
return entry, nil
}
// 设置任务运行模式(0: normal; 1: singleton; 2: once)
func (entry *Entry) SetMode(mode int) {
entry.mode.Set(mode)
// 是否单例运行
func (entry *Entry) IsSingleton() bool {
return entry.singleton.Val()
}
// 设置单例运行
func (entry *Entry) SetSingleton(enabled bool) {
entry.singleton.Set(enabled)
}
// 设置任务的运行次数
func (entry *Entry) SetTimes(times int) {
entry.times.Set(times)
}
// 定时任务状态
@ -54,6 +66,11 @@ func (entry *Entry) Status() int {
return entry.status.Val()
}
// 设置定时任务状态, 返回设置之前的状态
func (entry *Entry) SetStatus(status int) int {
return entry.status.Set(status)
}
// 启动定时任务
func (entry *Entry) Start() {
entry.status.Set(STATUS_READY)
@ -63,3 +80,8 @@ func (entry *Entry) Start() {
func (entry *Entry) Stop() {
entry.status.Set(STATUS_STOPPED)
}
// 关闭定时任务
func (entry *Entry) Close() {
entry.status.Set(STATUS_CLOSED)
}

110
g/os/gcron/gcron_unit_1_test.go
View File

@ -53,7 +53,7 @@ func TestCron_Add_Close(t *testing.T) {
})
}
func TestCron_Method(t *testing.T) {
func TestCron_Basic(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
cron.Add("* * * * * *", func() {}, "add")
@ -72,3 +72,111 @@ func TestCron_Method(t *testing.T) {
gtest.Assert(entry2, nil)
})
}
func TestCron_AddSingleton(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.AddSingleton("* * * * * *", func() {
array.Append(1)
time.Sleep(5*time.Second)
})
gtest.Assert(cron.Size(), 1)
time.Sleep(3500*time.Millisecond)
gtest.Assert(array.Len(), 1)
})
}
func TestCron_AddOnce(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.AddOnce("* * * * * *", func() {
array.Append(1)
})
cron.AddOnce("* * * * * *", func() {
array.Append(1)
})
gtest.Assert(cron.Size(), 2)
time.Sleep(2500*time.Millisecond)
gtest.Assert(array.Len(), 2)
gtest.Assert(cron.Size(), 0)
})
}
func TestCron_AddTimes(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.AddTimes("* * * * * *", 2, func() {
array.Append(1)
})
time.Sleep(3500*time.Millisecond)
gtest.Assert(array.Len(), 2)
gtest.Assert(cron.Size(), 0)
})
}
func TestCron_DelayAdd(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.DelayAdd(1, "* * * * * *", func() {
array.Append(1)
})
gtest.Assert(cron.Size(), 0)
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 0)
gtest.Assert(cron.Size(), 1)
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 1)
gtest.Assert(cron.Size(), 1)
})
}
func TestCron_DelayAddSingleton(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.DelayAddSingleton(1, "* * * * * *", func() {
array.Append(1)
time.Sleep(10*time.Second)
})
gtest.Assert(cron.Size(), 0)
time.Sleep(2200*time.Millisecond)
gtest.Assert(array.Len(), 1)
gtest.Assert(cron.Size(), 1)
})
}
func TestCron_DelayAddOnce(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.DelayAddOnce(1, "* * * * * *", func() {
array.Append(1)
})
gtest.Assert(cron.Size(), 0)
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 0)
gtest.Assert(cron.Size(), 1)
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 1)
gtest.Assert(cron.Size(), 0)
})
}
func TestCron_DelayAddTimes(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.DelayAddTimes(1, "* * * * * *", 2, func() {
array.Append(1)
})
gtest.Assert(cron.Size(), 0)
time.Sleep(5000*time.Millisecond)
gtest.Assert(array.Len(), 2)
gtest.Assert(cron.Size(), 0)
})
}

24
g/os/gcron/gcron_unit_2_test.go
View File

@ -4,6 +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 gcron_test
import (
@ -14,17 +15,30 @@ import (
"time"
)
func TestCron_AddSingleton(t *testing.T) {
func TestCron_Entry_Operations(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.AddSingleton("* * * * * *", func() {
entry, err1 := cron.Add("* * * * * *", func() {
array.Append(1)
time.Sleep(5*time.Second)
})
gtest.Assert(err1, nil)
gtest.Assert(array.Len(), 0)
gtest.Assert(cron.Size(), 1)
time.Sleep(3500*time.Millisecond)
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 1)
gtest.Assert(cron.Size(), 1)
entry.Stop()
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 1)
gtest.Assert(cron.Size(), 1)
entry.Start()
time.Sleep(1200*time.Millisecond)
gtest.Assert(array.Len(), 2)
gtest.Assert(cron.Size(), 1)
entry.Close()
time.Sleep(1200*time.Millisecond)
gtest.Assert(cron.Size(), 0)
})
}

32
g/os/gcron/gcron_unit_3_test.go
View File

@ -1,32 +0,0 @@
// Copyright 2018 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 gcron_test
import (
"gitee.com/johng/gf/g/container/garray"
"gitee.com/johng/gf/g/os/gcron"
"gitee.com/johng/gf/g/util/gtest"
"testing"
"time"
)
func TestCron_AddOnce(t *testing.T) {
gtest.Case(t, func() {
cron := gcron.New()
array := garray.New(0, 0)
cron.AddOnce("* * * * * *", func() {
array.Append(1)
})
cron.AddOnce("* * * * * *", func() {
array.Append(1)
})
gtest.Assert(cron.Size(), 2)
time.Sleep(2500*time.Millisecond)
gtest.Assert(array.Len(), 2)
gtest.Assert(cron.Size(), 0)
})
}