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

improve package gcron

Browse Source
This commit is contained in:
John Guo
2021-08-20 11:52:22 +08:00
parent a0ef6fce81
commit e10240bd7c
6 changed files with 204 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
os/gcron/gcron.go
View File

@ -48,63 +48,63 @@ func GetLogLevel() int {
}
// Add adds a timed task to default cron object.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func Add(pattern string, job func(), name ...string) (*Entry, error) {
return defaultCron.Add(pattern, job, name...)
}
// AddSingleton adds a singleton timed task, to default cron object.
// A singleton timed task is that can only be running one single instance at the same time.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func AddSingleton(pattern string, job func(), name ...string) (*Entry, error) {
return defaultCron.AddSingleton(pattern, job, name...)
}
// AddOnce adds a timed task which can be run only once, to default cron object.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func AddOnce(pattern string, job func(), name ...string) (*Entry, error) {
return defaultCron.AddOnce(pattern, job, name...)
}
// AddTimes adds a timed task which can be run specified times, to default cron object.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func AddTimes(pattern string, times int, job func(), name ...string) (*Entry, error) {
return defaultCron.AddTimes(pattern, times, job, name...)
}
// DelayAdd adds a timed task to default cron object after <delay> time.
// DelayAdd adds a timed task to default cron object after `delay` time.
func DelayAdd(delay time.Duration, pattern string, job func(), name ...string) {
defaultCron.DelayAdd(delay, pattern, job, name...)
}
// DelayAddSingleton adds a singleton timed task after <delay> time to default cron object.
// DelayAddSingleton adds a singleton timed task after `delay` time to default cron object.
func DelayAddSingleton(delay time.Duration, pattern string, job func(), name ...string) {
defaultCron.DelayAddSingleton(delay, pattern, job, name...)
}
// DelayAddOnce adds a timed task after <delay> time to default cron object.
// DelayAddOnce adds a timed task after `delay` time to default cron object.
// This timed task can be run only once.
func DelayAddOnce(delay time.Duration, pattern string, job func(), name ...string) {
defaultCron.DelayAddOnce(delay, pattern, job, name...)
}
// DelayAddTimes adds a timed task after <delay> time to default cron object.
// DelayAddTimes adds a timed task after `delay` time to default cron object.
// This timed task can be run specified times.
func DelayAddTimes(delay time.Duration, pattern string, times int, job func(), name ...string) {
defaultCron.DelayAddTimes(delay, pattern, times, job, name...)
}
// Search returns a scheduled task with the specified <name>.
// Search returns a scheduled task with the specified `name`.
// It returns nil if no found.
func Search(name string) *Entry {
return defaultCron.Search(name)
}
// Remove deletes scheduled task which named <name>.
// Remove deletes scheduled task which named `name`.
func Remove(name string) {
defaultCron.Remove(name)
}
@ -119,12 +119,12 @@ func Entries() []*Entry {
return defaultCron.Entries()
}
// Start starts running the specified timed task named <name>.
// Start starts running the specified timed task named `name`.
func Start(name string) {
defaultCron.Start(name)
}
// Stop stops running the specified timed task named <name>.
// Stop stops running the specified timed task named `name`.
func Stop(name string) {
defaultCron.Stop(name)
}

133
os/gcron/gcron_cron.go
View File

@ -7,7 +7,6 @@
package gcron
import (
"github.com/gogf/gf/errors/gerror"
"time"
"github.com/gogf/gf/container/garray"
@ -18,11 +17,18 @@ import (
)
type Cron struct {
idGen *gtype.Int64 // Used for unique name generation.
status *gtype.Int // Timed task status(0: Not Start; 1: Running; 2: Stopped; -1: Closed)
entries *gmap.StrAnyMap // All timed task entries.
logPath *gtype.String // Logging path(folder).
logLevel *gtype.Int // Logging level.
idGen *gtype.Int64 // Used for unique name generation.
status *gtype.Int // Timed task status(0: Not Start; 1: Running; 2: Stopped; -1: Closed)
entries *gmap.StrAnyMap // All timed task entries.
logger *glog.Logger // Logger, it is nil in default.
// Logging path(folder).
// Deprecated, use logger instead.
logPath *gtype.String
// Logging level.
// Deprecated, use logger instead.
logLevel *gtype.Int
}
// New returns a new Cron object with default settings.
@ -36,76 +42,101 @@ func New() *Cron {
}
}
// SetLogger sets the logger for cron.
func (c *Cron) SetLogger(logger *glog.Logger) {
c.logger = logger
}
// GetLogger returns the logger in the cron.
func (c *Cron) GetLogger() *glog.Logger {
return c.logger
}
// SetLogPath sets the logging folder path.
// Deprecated, use SetLogger instead.
func (c *Cron) SetLogPath(path string) {
c.logPath.Set(path)
}
// GetLogPath return the logging folder path.
// Deprecated, use GetLogger instead.
func (c *Cron) GetLogPath() string {
return c.logPath.Val()
}
// SetLogLevel sets the logging level.
// Deprecated, use SetLogger instead.
func (c *Cron) SetLogLevel(level int) {
c.logLevel.Set(level)
}
// GetLogLevel returns the logging level.
// Deprecated, use GetLogger instead.
func (c *Cron) GetLogLevel() int {
return c.logLevel.Val()
}
// Add adds a timed task.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
func (c *Cron) Add(pattern string, job func(), name ...string) (*Entry, error) {
// AddEntry creates and returns a new Entry object.
func (c *Cron) AddEntry(pattern string, job func(), times int, singleton bool, name ...string) (*Entry, error) {
var (
entryName = ""
infinite = false
)
if len(name) > 0 {
if c.Search(name[0]) != nil {
return nil, gerror.NewCodef(gerror.CodeInvalidOperation, `cron job "%s" already exists`, name[0])
}
entryName = name[0]
}
return c.addEntry(pattern, job, false, name...)
if times <= 0 {
infinite = true
}
return c.doAddEntry(addEntryInput{
Name: entryName,
Job: job,
Times: times,
Pattern: pattern,
Singleton: singleton,
Infinite: infinite,
})
}
// Add adds a timed task.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func (c *Cron) Add(pattern string, job func(), name ...string) (*Entry, error) {
return c.AddEntry(pattern, job, -1, false, name...)
}
// AddSingleton adds a singleton timed task.
// A singleton timed task is that can only be running one single instance at the same time.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func (c *Cron) AddSingleton(pattern string, job func(), name ...string) (*Entry, error) {
if entry, err := c.Add(pattern, job, name...); err != nil {
return nil, err
} else {
entry.SetSingleton(true)
return entry, nil
}
}
// AddOnce adds a timed task which can be run only once.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
func (c *Cron) AddOnce(pattern string, job func(), name ...string) (*Entry, error) {
if entry, err := c.Add(pattern, job, name...); err != nil {
return nil, err
} else {
entry.SetTimes(1)
return entry, nil
}
return c.AddEntry(pattern, job, -1, true, name...)
}
// AddTimes adds a timed task which can be run specified times.
// A unique <name> can be bound with the timed task.
// It returns and error if the <name> is already used.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
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
}
return c.AddEntry(pattern, job, times, false, name...)
}
// DelayAdd adds a timed task after <delay> time.
// AddOnce adds a timed task which can be run only once.
// A unique `name` can be bound with the timed task.
// It returns and error if the `name` is already used.
func (c *Cron) AddOnce(pattern string, job func(), name ...string) (*Entry, error) {
return c.AddEntry(pattern, job, 1, false, name...)
}
// DelayAddEntry adds a timed task after `delay` time.
func (c *Cron) DelayAddEntry(delay time.Duration, pattern string, job func(), times int, singleton bool, name ...string) {
gtimer.AddOnce(delay, func() {
if _, err := c.AddEntry(pattern, job, times, singleton, name...); err != nil {
panic(err)
}
})
}
// DelayAdd adds a timed task after `delay` time.
func (c *Cron) DelayAdd(delay time.Duration, pattern string, job func(), name ...string) {
gtimer.AddOnce(delay, func() {
if _, err := c.Add(pattern, job, name...); err != nil {
@ -114,7 +145,7 @@ func (c *Cron) DelayAdd(delay time.Duration, pattern string, job func(), name ..
})
}
// DelayAddSingleton adds a singleton timed task after <delay> time.
// DelayAddSingleton adds a singleton timed task after `delay` time.
func (c *Cron) DelayAddSingleton(delay time.Duration, pattern string, job func(), name ...string) {
gtimer.AddOnce(delay, func() {
if _, err := c.AddSingleton(pattern, job, name...); err != nil {
@ -123,7 +154,7 @@ func (c *Cron) DelayAddSingleton(delay time.Duration, pattern string, job func()
})
}
// DelayAddOnce adds a timed task after <delay> time.
// DelayAddOnce adds a timed task after `delay` time.
// This timed task can be run only once.
func (c *Cron) DelayAddOnce(delay time.Duration, pattern string, job func(), name ...string) {
gtimer.AddOnce(delay, func() {
@ -133,7 +164,7 @@ func (c *Cron) DelayAddOnce(delay time.Duration, pattern string, job func(), nam
})
}
// DelayAddTimes adds a timed task after <delay> time.
// DelayAddTimes adds a timed task after `delay` time.
// This timed task can be run specified times.
func (c *Cron) DelayAddTimes(delay time.Duration, pattern string, times int, job func(), name ...string) {
gtimer.AddOnce(delay, func() {
@ -143,8 +174,8 @@ func (c *Cron) DelayAddTimes(delay time.Duration, pattern string, times int, job
})
}
// Search returns a scheduled task with the specified <name>.
// It returns nil if no found.
// Search returns a scheduled task with the specified `name`.
// It returns nil if not found.
func (c *Cron) Search(name string) *Entry {
if v := c.entries.Get(name); v != nil {
return v.(*Entry)
@ -152,7 +183,7 @@ func (c *Cron) Search(name string) *Entry {
return nil
}
// Start starts running the specified timed task named <name>.
// Start starts running the specified timed task named `name`.
func (c *Cron) Start(name ...string) {
if len(name) > 0 {
for _, v := range name {
@ -165,7 +196,7 @@ func (c *Cron) Start(name ...string) {
}
}
// Stop stops running the specified timed task named <name>.
// Stop stops running the specified timed task named `name`.
func (c *Cron) Stop(name ...string) {
if len(name) > 0 {
for _, v := range name {
@ -178,7 +209,7 @@ func (c *Cron) Stop(name ...string) {
}
}
// Remove deletes scheduled task which named <name>.
// Remove deletes scheduled task which named `name`.
func (c *Cron) Remove(name string) {
if v := c.entries.Get(name); v != nil {
v.(*Entry).Close()

127
os/gcron/gcron_entry.go
View File

@ -7,117 +7,131 @@
package gcron
import (
"github.com/gogf/gf/errors/gerror"
"reflect"
"runtime"
"time"
"github.com/gogf/gf/container/gtype"
"github.com/gogf/gf/os/glog"
"github.com/gogf/gf/os/gtimer"
"github.com/gogf/gf/util/gconv"
)
// Entry is timing task entry.
type Entry struct {
cron *Cron // Cron object belonged to.
entry *gtimer.Entry // Associated gtimer.Entry.
schedule *cronSchedule // Timed schedule object.
jobName string // Callback function name(address info).
times *gtype.Int // Running times limit.
Name string // Entry name.
Job func() `json:"-"` // Callback function.
Time time.Time // Registered time.
cron *Cron // Cron object belonged to.
timerEntry *gtimer.Entry // Associated timer Entry.
schedule *cronSchedule // Timed schedule object.
jobName string // Callback function name(address info).
times *gtype.Int // Running times limit.
infinite *gtype.Bool // No times limit.
Name string // Entry name.
Job func() `json:"-"` // Callback function.
Time time.Time // Registered time.
}
// addEntry creates and returns a new Entry object.
// Param <job> is the callback function for timed task execution.
// Param <singleton> specifies whether timed task executing in singleton mode.
// Param <name> names this entry for manual control.
func (c *Cron) addEntry(pattern string, job func(), singleton bool, name ...string) (*Entry, error) {
schedule, err := newSchedule(pattern)
type addEntryInput struct {
Name string // Name names this entry for manual control.
Job func() // Job is the callback function for timed task execution.
Times int // Times specifies the running limit times for the entry.
Pattern string // Pattern is the crontab style string for scheduler.
Singleton bool // Singleton specifies whether timed task executing in singleton mode.
Infinite bool // Infinite specifies whether this entry is running with no times limit.
}
// doAddEntry creates and returns a new Entry object.
func (c *Cron) doAddEntry(in addEntryInput) (*Entry, error) {
if in.Name != "" {
if c.Search(in.Name) != nil {
return nil, gerror.NewCodef(gerror.CodeInvalidOperation, `cron job "%s" already exists`, in.Name)
}
}
schedule, err := newSchedule(in.Pattern)
if err != nil {
return nil, err
}
// No limit for <times>, for gtimer checking scheduling every second.
// No limit for `times`, for timer checking scheduling every second.
entry := &Entry{
cron: c,
schedule: schedule,
jobName: runtime.FuncForPC(reflect.ValueOf(job).Pointer()).Name(),
times: gtype.NewInt(defaultTimes),
Job: job,
jobName: runtime.FuncForPC(reflect.ValueOf(in.Job).Pointer()).Name(),
times: gtype.NewInt(in.Times),
infinite: gtype.NewBool(in.Infinite),
Job: in.Job,
Time: time.Now(),
}
if len(name) > 0 {
entry.Name = name[0]
if in.Name != "" {
entry.Name = in.Name
} else {
entry.Name = "gcron-" + gconv.String(c.idGen.Add(1))
entry.Name = "cron-" + gconv.String(c.idGen.Add(1))
}
// When you add a scheduled task, you cannot allow it to run.
// It cannot start running when added to gtimer.
// It cannot start running when added to timer.
// It should start running after the entry is added to the Cron entries map, to avoid the task
// from running during adding where the entries do not have the entry information, which might cause panic.
entry.entry = gtimer.AddEntry(time.Second, entry.check, singleton, -1, gtimer.StatusStopped)
entry.timerEntry = gtimer.AddEntry(time.Second, entry.check, in.Singleton, -1, gtimer.StatusStopped)
c.entries.Set(entry.Name, entry)
entry.entry.Start()
entry.timerEntry.Start()
return entry, nil
}
// IsSingleton return whether this entry is a singleton timed task.
func (entry *Entry) IsSingleton() bool {
return entry.entry.IsSingleton()
return entry.timerEntry.IsSingleton()
}
// SetSingleton sets the entry running in singleton mode.
func (entry *Entry) SetSingleton(enabled bool) {
entry.entry.SetSingleton(enabled)
entry.timerEntry.SetSingleton(enabled)
}
// SetTimes sets the times which the entry can run.
func (entry *Entry) SetTimes(times int) {
entry.times.Set(times)
entry.infinite.Set(false)
}
// Status returns the status of entry.
func (entry *Entry) Status() int {
return entry.entry.Status()
return entry.timerEntry.Status()
}
// SetStatus sets the status of the entry.
func (entry *Entry) SetStatus(status int) int {
return entry.entry.SetStatus(status)
return entry.timerEntry.SetStatus(status)
}
// Start starts running the entry.
func (entry *Entry) Start() {
entry.entry.Start()
entry.timerEntry.Start()
}
// Stop stops running the entry.
func (entry *Entry) Stop() {
entry.entry.Stop()
entry.timerEntry.Stop()
}
// Close stops and removes the entry from cron.
func (entry *Entry) Close() {
entry.cron.entries.Remove(entry.Name)
entry.entry.Close()
entry.timerEntry.Close()
}
// Timing task check execution.
// check is the core timing task check logic.
// The running times limits feature is implemented by gcron.Entry and cannot be implemented by gtimer.Entry.
// gcron.Entry relies on gtimer to implement a scheduled task check for gcron.Entry per second.
func (entry *Entry) check() {
if entry.schedule.meet(time.Now()) {
var (
path = entry.cron.GetLogPath()
level = entry.cron.GetLogLevel()
)
switch entry.cron.status.Val() {
case StatusStopped:
return
case StatusClosed:
glog.Path(path).Level(level).Debugf("[gcron] %s(%s) %s removed", entry.Name, entry.schedule.pattern, entry.jobName)
entry.logDebugf(
"[gcron] %s(%s) %s removed",
entry.Name, entry.schedule.pattern, entry.jobName,
)
entry.Close()
case StatusReady:
@ -125,35 +139,48 @@ func (entry *Entry) check() {
case StatusRunning:
defer func() {
if err := recover(); err != nil {
glog.Path(path).Level(level).Errorf(
entry.logErrorf(
"[gcron] %s(%s) %s end with error: %+v",
entry.Name, entry.schedule.pattern, entry.jobName, err,
)
} else {
glog.Path(path).Level(level).Debugf(
entry.logDebugf(
"[gcron] %s(%s) %s end",
entry.Name, entry.schedule.pattern, entry.jobName,
)
}
if entry.entry.Status() == StatusClosed {
if entry.timerEntry.Status() == StatusClosed {
entry.Close()
}
}()
// Running times check.
times := entry.times.Add(-1)
if times <= 0 {
if entry.entry.SetStatus(StatusClosed) == StatusClosed || times < 0 {
return
if !entry.infinite.Val() {
times := entry.times.Add(-1)
if times <= 0 {
if entry.timerEntry.SetStatus(StatusClosed) == StatusClosed || times < 0 {
return
}
}
}
if times < 2000000000 && times > 1000000000 {
entry.times.Set(defaultTimes)
}
glog.Path(path).Level(level).Debugf("[gcron] %s(%s) %s start", entry.Name, entry.schedule.pattern, entry.jobName)
entry.logDebugf(
"[gcron] %s(%s) %s start",
entry.Name, entry.schedule.pattern, entry.jobName,
)
entry.Job()
}
}
}
func (entry *Entry) logDebugf(format string, v ...interface{}) {
if logger := entry.cron.GetLogger(); logger != nil {
logger.Debugf(format, v...)
}
}
func (entry *Entry) logErrorf(format string, v ...interface{}) {
if logger := entry.cron.GetLogger(); logger != nil {
logger.Errorf(format, v...)
}
}

2
os/gcron/gcron_schedule.go
View File

@ -223,7 +223,7 @@ func parseItemValue(value string, fieldType byte) (int, error) {
return 0, gerror.NewCodef(gerror.CodeInvalidParameter, `invalid pattern value: "%s"`, value)
}
// meet checks if the given time <t> meets the runnable point for the job.
// meet checks if the given time `t` meets the runnable point for the job.
func (s *cronSchedule) meet(t time.Time) bool {
if s.every != 0 {
// It checks using interval.

23
os/gtimer/gtimer_entry.go
View File

@ -8,18 +8,18 @@ package gtimer
import (
"github.com/gogf/gf/container/gtype"
"math"
)
// Entry is the timing job.
type Entry struct {
job JobFunc // The job function.
timer *Timer // Belonged timer.
ticks int64 // The job runs every ticks.
ticks int64 // The job runs every tick.
times *gtype.Int // Limit running times.
status *gtype.Int // Job status.
singleton *gtype.Bool // Singleton mode.
nextTicks *gtype.Int64 // Next run ticks of the job.
infinite *gtype.Bool // No times limit.
}
// JobFunc is the job function.
@ -32,15 +32,13 @@ func (entry *Entry) Status() int {
// Run runs the timer job asynchronously.
func (entry *Entry) Run() {
leftRunningTimes := entry.times.Add(-1)
// It checks its running times exceeding.
if leftRunningTimes < 0 {
entry.status.Set(StatusClosed)
return
}
// This means it has no limit in running times.
if leftRunningTimes == math.MaxInt32-1 {
entry.times.Set(math.MaxInt32)
if !entry.infinite.Val() {
leftRunningTimes := entry.times.Add(-1)
// It checks its running times exceeding.
if leftRunningTimes < 0 {
entry.status.Set(StatusClosed)
return
}
}
go func() {
defer func() {
@ -108,7 +106,7 @@ func (entry *Entry) Close() {
entry.status.Set(StatusClosed)
}
// Reset reset the job, which resets its ticks for next running.
// Reset resets the job, which resets its ticks for next running.
func (entry *Entry) Reset() {
entry.nextTicks.Set(entry.timer.ticks.Val() + entry.ticks)
}
@ -131,4 +129,5 @@ func (entry *Entry) Job() JobFunc {
// SetTimes sets the limit running times for the job.
func (entry *Entry) SetTimes(times int) {
entry.times.Set(times)
entry.infinite.Set(false)
}

28
os/gtimer/gtimer_timer.go
View File

@ -118,8 +118,11 @@ func (t *Timer) Close() {
// createEntry creates and adds a timing job to the timer.
func (t *Timer) createEntry(interval time.Duration, job JobFunc, singleton bool, times int, status int) *Entry {
var (
infinite = false
)
if times <= 0 {
times = defaultTimes
infinite = true
}
var (
intervalTicksOfJob = int64(interval / t.options.Interval)
@ -129,16 +132,19 @@ func (t *Timer) createEntry(interval time.Duration, job JobFunc, singleton bool,
// then sets it to one tick, which means it will be run in one interval.
intervalTicksOfJob = 1
}
nextTicks := t.ticks.Val() + intervalTicksOfJob
entry := &Entry{
job: job,
timer: t,
ticks: intervalTicksOfJob,
times: gtype.NewInt(times),
status: gtype.NewInt(status),
singleton: gtype.NewBool(singleton),
nextTicks: gtype.NewInt64(nextTicks),
}
var (
nextTicks = t.ticks.Val() + intervalTicksOfJob
entry = &Entry{
job: job,
timer: t,
ticks: intervalTicksOfJob,
times: gtype.NewInt(times),
status: gtype.NewInt(status),
singleton: gtype.NewBool(singleton),
nextTicks: gtype.NewInt64(nextTicks),
infinite: gtype.NewBool(infinite),
}
)
t.queue.Push(entry, nextTicks)
return entry
}