diff --git a/cluster.go b/cluster.go index e93d46e..774948d 100644 --- a/cluster.go +++ b/cluster.go @@ -631,7 +631,7 @@ func (l *Cluster) processTask(taskId string) { l.logger.Errorf(ctx, "doTask timer:获取锁失败:%s", taskId) return } - if b, err := lock.Lock(); !b { + if b, err := lock.Lock(); !b || err != nil { l.logger.Errorf(ctx, "doTask timer:获取锁失败:%s %+v", taskId, err) return } diff --git a/next_time.go b/next_time.go index a5326bf..06e0126 100644 --- a/next_time.go +++ b/next_time.go @@ -119,9 +119,6 @@ func calculateNextInterval(t time.Time, job JobData) (*time.Time, error) { // 计算下一个执行时间 next := job.BaseTime.Add((intervals + 1) * job.IntervalTime) - // 需要整的 - next = next.Round(job.IntervalTime) - // 确保下次执行时间不早于当前时间 if next.Before(t) || next.Equal(t) { next = next.Add(job.IntervalTime) diff --git a/next_time_test.go b/next_time_test.go index cd5d4a0..d0bdb50 100644 --- a/next_time_test.go +++ b/next_time_test.go @@ -90,7 +90,7 @@ func TestGetNextTime(t *testing.T) { BaseTime: tt, IntervalTime: 1 * time.Hour, }, - expectedTime: time.Date(2025, 10, 16, 12, 00, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM + expectedTime: tt.Add(1 * time.Hour), expectedError: nil, }, { @@ -100,7 +100,7 @@ func TestGetNextTime(t *testing.T) { BaseTime: tt, IntervalTime: 1 * time.Minute, }, - expectedTime: time.Date(2025, 10, 16, 10, 31, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM + expectedTime: tt.Add(1 * time.Minute), expectedError: nil, }, { diff --git a/once.go b/once.go index 571af20..195e731 100644 --- a/once.go +++ b/once.go @@ -106,6 +106,11 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c return nil, errors.New("redis client is nil") } + if call == nil { + op.logger.Errorf(ctx, "worker callback is nil") + return nil, errors.New("worker callback is nil") + } + // 清理 keyPrefix 中的 {} 字符，避免破坏 Redis Cluster hash tag keyPrefix = sanitizeKeyPrefix(keyPrefix) @@ -477,14 +482,14 @@ func (l *Once) create(ctx context.Context, source createSource, jobType jobType, l.logger.Errorf(ctx, "processTask timer:获取锁失败:%s", taskId) return err } - if b, err := lock.Lock(); !b { + if b, err := lock.Lock(); !b || err != nil { l.logger.Errorf(ctx, "processTask timer:获取锁失败:%s %+v", taskId, err) return err } defer lock.Unlock() } - score, err := l.redis.ZScore(l.ctx, l.zsetKey, redisKey).Result() + _, err := l.redis.ZScore(l.ctx, l.zsetKey, redisKey).Result() if err != nil { if errors.Is(err, redis.Nil) { return l.save(ctx, jobType, taskType, taskId, taskTimes, attachData, runCount) @@ -492,12 +497,8 @@ func (l *Once) create(ctx context.Context, source createSource, jobType jobType, l.logger.Errorf(l.ctx, "redis.ZScore err:%v", err) return err } - if score > 0 { l.logger.Errorf(l.ctx, "task exists taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount) return ErrTaskExists - } - - return l.save(ctx, jobType, taskType, taskId, taskTimes, attachData, runCount) } // 删除任务 @@ -600,7 +601,7 @@ func (l *Once) processTask(key string) { l.logger.Errorf(ctx, "processTask timer:获取锁失败:%s", taskId) return } - if b, err := lock.Lock(); !b { + if b, err := lock.Lock(); !b || err != nil { l.logger.Errorf(ctx, "processTask timer:获取锁失败:%s %+v", taskId, err) return } @@ -643,7 +644,7 @@ func (l *Once) processTask(key string) { // 删除任务 l.logger.Infof(ctx, "processTask delete key:%s", key) if err := l.Delete(taskType, taskId); err != nil { - l.logger.Errorf(ctx, "processTask delete errprocessTask delete err:%v", err) + l.logger.Errorf(ctx, "processTask delete err:%v", err) } return } diff --git a/option.go b/option.go index 9fd2aef..a245132 100644 --- a/option.go +++ b/option.go @@ -122,7 +122,7 @@ func WithMaxRetryCount(count int) Option { func WithMaxWorkers(count int) Option { return func(o *Options) { - if count < 0 { + if count <= 0 { count = 100 } o.maxWorkers = count diff --git a/single.go b/single.go index 01a6333..93629d1 100644 --- a/single.go +++ b/single.go @@ -1,504 +1,508 @@ -package timerx - -// 作者：黄新云 - -import ( - "context" - "fmt" - "runtime/debug" - "sync" - "sync/atomic" - "time" - - "github.com/google/uuid" - "github.com/robfig/cron/v3" - "github.com/yuninks/timerx/logger" -) - -// 简单定时器 -// 1. 这个定时器的作用范围是本机 -// 2. 适用简单的时间间隔定时任务 - -type Single struct { - ctx context.Context - cancel context.CancelFunc - logger logger.Logger - location *time.Location - nextTime time.Time - nextTimeMux sync.RWMutex - wg sync.WaitGroup - workerList sync.Map // 任务列表，key为taskId，value为worker - timerIndex int64 // 任务索引，用于生成taskId - stopChan chan struct{} // 停止信号 - hasRun sync.Map // 记录已经执行的任务，key为taskId，value为执行时间 - timeout time.Duration // 单次任务超时时间 - cronParser *cron.Parser // cron表达式解析器 -} - -// 定时器类 -// @param ctx context.Context 上下文 -// @param opts ...Option 配置项 -func InitSingle(ctx context.Context, opts ...Option) *Single { - op := newOptions(opts...) - ctx, cancel := context.WithCancel(ctx) - - sin := &Single{ - ctx: ctx, - cancel: cancel, - logger: op.logger, - location: op.location, - nextTime: time.Now(), - stopChan: make(chan struct{}), - timeout: op.timeout, - cronParser: op.cronParser, - } - - sin.startDaemon() - - return sin -} - -func (l *Single) startDaemon() { - - l.wg.Add(1) - go l.timerLoop() - - l.wg.Add(1) - go l.cleanupLoop() - -} - -// 停止所有定时任务 -func (l *Single) Stop() { - close(l.stopChan) - - if l.cancel != nil { - l.cancel() - } - - l.wg.Wait() - l.logger.Infof(l.ctx, "timer single: stopped") -} - -// 获取任务数量 -func (l *Single) TaskCount() int { - count := 0 - l.workerList.Range(func(k, v interface{}) bool { - count++ - return true - }) - return count -} - -func (l *Single) MaxIndex() int64 { - return atomic.LoadInt64(&l.timerIndex) + 1 -} - -// 定时器主循环 -func (l *Single) timerLoop() { - defer l.wg.Done() - - ticker := time.NewTicker(100 * time.Millisecond) // 提高精度到100ms - defer ticker.Stop() - - for { - select { - case t := <-ticker.C: - l.nextTimeMux.RLock() - nextTime := l.nextTime - l.nextTimeMux.RUnlock() - - if t.Before(nextTime) { - continue - } - - l.iterator(l.ctx) - - case <-l.ctx.Done(): - l.logger.Infof(l.ctx, "timer: context cancelled, stopping timer loop") - return - case <-l.stopChan: - l.logger.Infof(l.ctx, "timer: received stop signal, stopping timer loop") - return - } - } -} - -// 清理循环 -func (s *Single) cleanupLoop() { - defer s.wg.Done() - - ticker := time.NewTicker(time.Minute) - defer ticker.Stop() - - for { - select { - case <-ticker.C: - now := time.Now() - cleanupTime := now.Add(-2 * time.Minute) // 清理2分钟前的记录 - - s.hasRun.Range(func(k, v any) bool { - t, ok := v.(time.Time) - if !ok || t.Before(cleanupTime) { - s.hasRun.Delete(k) - } - return true - }) - - case <-s.ctx.Done(): - s.logger.Infof(s.ctx, "timer: context cancelled, stopping cleanup loop") - return - case <-s.stopChan: - s.logger.Infof(s.ctx, "timer: received stop signal, stopping cleanup loop") - return - } - } -} - -// 每月执行一次 -// @param ctx 上下文 -// @param taskId 任务ID -// @param day 每月的几号 -// @param hour 小时 -// @param minute 分钟 -// @param second 秒 -// @param callback 回调函数 -// @param extendData 扩展数据 -// @return error -func (c *Single) EveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { - - // nowTime := time.Now().In(c.location) - - jobData := JobData{ - JobType: JobTypeEveryMonth, - TaskId: taskId, - // CreateTime: nowTime, - Day: day, - Hour: hour, - Minute: minute, - Second: second, - } - - return c.addJob(ctx, jobData, callback, extendData) -} - -// 每周执行一次 -// @param ctx context.Context 上下文 -// @param taskId string 任务ID -// @param week time.Weekday 周 -// @param hour int 小时 -// @param minute int 分钟 -// @param second int 秒 -func (c *Single) EveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { - // nowTime := time.Now().In(c.location) - - jobData := JobData{ - JobType: JobTypeEveryWeek, - TaskId: taskId, - // CreateTime: nowTime, - Weekday: week, - Hour: hour, - Minute: minute, - Second: second, - } - - return c.addJob(ctx, jobData, callback, extendData) -} - -// 每天执行一次 -func (c *Single) EveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { - // nowTime := time.Now().In(c.location) - - jobData := JobData{ - JobType: JobTypeEveryDay, - TaskId: taskId, - // CreateTime: nowTime, - Hour: hour, - Minute: minute, - Second: second, - } - - return c.addJob(ctx, jobData, callback, extendData) -} - -// 每小时执行一次 -func (c *Single) EveryHour(ctx context.Context, taskId string, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { - // nowTime := time.Now().In(c.location) - - jobData := JobData{ - JobType: JobTypeEveryHour, - TaskId: taskId, - // CreateTime: nowTime, - Minute: minute, - Second: second, - } - - return c.addJob(ctx, jobData, callback, extendData) -} - -// 每分钟执行一次 -func (c *Single) EveryMinute(ctx context.Context, taskId string, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { - // nowTime := time.Now().In(c.location) - - jobData := JobData{ - JobType: JobTypeEveryMinute, - TaskId: taskId, - // CreateTime: nowTime, - Second: second, - } - - return c.addJob(ctx, jobData, callback, extendData) -} - -// 特定时间间隔 -func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { - - if spaceTime < 0 { - c.logger.Errorf(ctx, "间隔时间不能小于0") - return 0, ErrIntervalTime - } - - // 固定时间点为20250101 00:00:00，便于计算下一次执行时间 - zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, c.location) - - jobData := JobData{ - JobType: JobTypeInterval, - TaskId: taskId, - // CreateTime: nowTime, - BaseTime: zeroTime, - IntervalTime: spaceTime, - } - - return c.addJob(ctx, jobData, callback, extendData) -} - -func (l *Single) Cron(ctx context.Context, taskId string, cronExpression string, callback func(ctx context.Context, extendData any) error, extendData any, opt ...Option) (int64, error) { - // 固定时间点为20250101 00:00:00，便于计算下一次执行时间 - zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, l.location) - - options := Options{} - for _, o := range opt { - o(&options) - } - cronParser := l.cronParser - if options.cronParser != nil { - cronParser = options.cronParser - } - - sche, err := GetCronSche(cronExpression, cronParser) - if err != nil { - l.logger.Errorf(ctx, "timer Single Cron cronExpression error:%s", err.Error()) - return 0, err - } - - jobData := JobData{ - JobType: JobTypeCron, - TaskId: taskId, - BaseTime: zeroTime, // 默认当天的零点 - CronExpression: cronExpression, - CronSchedule: sche, - } - return l.addJob(ctx, jobData, callback, extendData) -} - -// 间隔定时器 -// @param space 间隔时间 -// @param call 回调函数 -// @param extend 附加参数 -// @return int 定时器索引 -// @return error 错误 -func (l *Single) addJob(ctx context.Context, jobData JobData, call func(ctx context.Context, extendData interface{}) error, extend interface{}) (int64, error) { - if jobData.TaskId == "" { - l.logger.Errorf(ctx, "任务ID不能为空") - return 0, ErrTaskIdEmpty - } - if jobData.Day < 0 || jobData.Day > 31 { - l.logger.Errorf(ctx, "每月的天数必须在0-31之间") - return 0, ErrMonthDay - } - if jobData.Hour < 0 || jobData.Hour > 23 { - l.logger.Errorf(ctx, "小时必须在0-23之间") - return 0, ErrHour - } - if jobData.Minute < 0 || jobData.Minute > 59 { - l.logger.Errorf(ctx, "分钟必须在0-59之间") - return 0, ErrMinute - } - if jobData.Second < 0 || jobData.Second > 59 { - l.logger.Errorf(ctx, "秒必须在0-59之间") - return 0, ErrSecond - } - if call == nil { - l.logger.Errorf(ctx, "回调函数不能为空") - return 0, ErrCallbackEmpty - } - - nextTime, err := GetNextTime(time.Now().In(l.location), jobData) - if err != nil { - l.logger.Errorf(ctx, "获取下次执行时间失败:%s", err.Error()) - return 0, err - } - jobData.NextTime = *nextTime - - // 生成唯一索引 - index := atomic.AddInt64(&l.timerIndex, 1) - - t := timerStr{ - Callback: call, - CanRunning: make(chan struct{}, 1), - ExtendData: extend, - TaskId: jobData.TaskId, - JobData: &jobData, - } - - l.workerList.Store(index, t) - - // 计算下次执行时间(全局) - l.updateNextTimeIfEarlier(*nextTime) - - return index, nil -} - -// 如果更早则更新下次执行时间(全局) -func (s *Single) updateNextTimeIfEarlier(candidate time.Time) { - s.nextTimeMux.Lock() - defer s.nextTimeMux.Unlock() - - if candidate.Before(s.nextTime) { - s.nextTime = candidate - } -} - -// 删除定时器 -func (l *Single) Del(index int64) { - if _, ok := l.workerList.Load(index); ok { - l.workerList.Delete(index) - } -} - -func (l *Single) DelByTaskId(taskId string) { - l.workerList.Range(func(k, v interface{}) bool { - timeStr, ok := v.(timerStr) - if ok && timeStr.TaskId == taskId { - l.workerList.Delete(k) - } - return true - }) -} - -// 迭代定时器列表 -func (l *Single) iterator(ctx context.Context) { - // 当前时间 - nowTime := time.Now().In(l.location) - - // 默认5秒后（如果没有值就暂停进来5秒） - newNextTime := nowTime.Add(time.Second * 5) - - l.workerList.Range(func(k, v interface{}) bool { - timeStr, ok := v.(timerStr) - if !ok { - l.logger.Errorf(ctx, "timer: 类型断言失败，跳过该任务") - l.workerList.Delete(k) - return true - } - - if timeStr.JobData.NextTime.Before(nowTime) || timeStr.JobData.NextTime.Equal(nowTime) { - - originTime := timeStr.JobData.NextTime - - // 计算下次执行时间 - nextTime, err := GetNextTime(nowTime, *timeStr.JobData) - if err != nil { - l.logger.Errorf(ctx, "timer: 计算下次执行时间失败:%s", err.Error()) - return true - } - // 更新下次执行时间 - timeStr.JobData.NextTime = *nextTime - - if nextTime.Before(newNextTime) { - // 本规则下次发送时间小于系统下次需要执行的时间：替换 - newNextTime = *nextTime - } - - go l.executeTask(ctx, timeStr, originTime) - - } - - return true - - }) - - l.updateNextTime(newNextTime) -} - -// 执行任务 -func (s *Single) executeTask(ctx context.Context, timer timerStr, originTime time.Time) { - // 创建带追踪ID的上下文 - - u, _ := uuid.NewV7() - - traceCtx := context.WithValue(ctx, "trace_id", u.String()) - s.logger.Infof(traceCtx, "timer Single begin taskId:%s originTime:%d", timer.TaskId, originTime.UnixMilli()) - traceCtx, cancel := context.WithTimeout(traceCtx, s.timeout) // 设置执行超时 - defer cancel() - - select { - case timer.CanRunning <- struct{}{}: - defer func() { - select { - case <-timer.CanRunning: - default: - } - }() - - // 执行回调 - begin := time.Now() - if err := s.doTask(traceCtx, timer, originTime); err != nil { - s.logger.Errorf(traceCtx, "timer: 任务执行失败: %s", err.Error()) - } - s.logger.Infof(traceCtx, "timer Single end taskId:%s originTime:%d cost:%dms", timer.TaskId, originTime.UnixMilli(), time.Since(begin).Milliseconds()) - case <-traceCtx.Done(): - s.logger.Errorf(traceCtx, "timer: 任务执行超时: %s", timer.TaskId) - default: - // 任务正在执行中，跳过本次 - s.logger.Infof(traceCtx, "timer: 任务正在执行中，跳过本次 %s", timer.TaskId) - } -} - -// 定时器操作类 -// 这里不应painc -func (l *Single) doTask(ctx context.Context, timeStr timerStr, originTime time.Time) error { - // 检查任务是否已执行 - taskKey := fmt.Sprintf("%s:%d", timeStr.TaskId, originTime.UnixMilli()) - if _, loaded := l.hasRun.LoadOrStore(taskKey, time.Now()); loaded { - l.logger.Errorf(ctx, "timer: 任务已执行，跳过本次执行 %s", timeStr.TaskId) - return ErrTaskExecuted - } - - defer func() { - if err := recover(); err != nil { - l.logger.Errorf(ctx, "timer Single call panic err:%+v stack:%s", err, string(debug.Stack())) - } - }() - - err := timeStr.Callback(ctx, timeStr.ExtendData) - if err != nil { - l.logger.Errorf(ctx, "timer Single call back %s, err: %v", timeStr.TaskId, err) - return err - } - - return nil -} - -// 更新下次执行时间 -func (s *Single) updateNextTime(newTime time.Time) { - s.nextTimeMux.Lock() - defer s.nextTimeMux.Unlock() - - now := time.Now() - if newTime.Before(now) { - s.nextTime = now - } else { - s.nextTime = newTime - } -} +package timerx + +// 作者：黄新云 + +import ( + "context" + "fmt" + "runtime/debug" + "sync" + "sync/atomic" + "time" + + "github.com/google/uuid" + "github.com/robfig/cron/v3" + "github.com/yuninks/timerx/logger" +) + +// 简单定时器 +// 1. 这个定时器的作用范围是本机 +// 2. 适用简单的时间间隔定时任务 + +type Single struct { + ctx context.Context + cancel context.CancelFunc + logger logger.Logger + location *time.Location + nextTime time.Time + nextTimeMux sync.RWMutex + wg sync.WaitGroup + workerList sync.Map // 任务列表，key为taskId，value为worker + timerIndex int64 // 任务索引，用于生成taskId + stopChan chan struct{} // 停止信号 + hasRun sync.Map // 记录已经执行的任务，key为taskId，value为执行时间 + timeout time.Duration // 单次任务超时时间 + cronParser *cron.Parser // cron表达式解析器 +} + +// 定时器类 +// @param ctx context.Context 上下文 +// @param opts ...Option 配置项 +func InitSingle(ctx context.Context, opts ...Option) *Single { + op := newOptions(opts...) + ctx, cancel := context.WithCancel(ctx) + + sin := &Single{ + ctx: ctx, + cancel: cancel, + logger: op.logger, + location: op.location, + nextTime: time.Now(), + stopChan: make(chan struct{}), + timeout: op.timeout, + cronParser: op.cronParser, + } + + sin.startDaemon() + + return sin +} + +func (l *Single) startDaemon() { + + l.wg.Add(1) + go l.timerLoop() + + l.wg.Add(1) + go l.cleanupLoop() + +} + +// 停止所有定时任务 +func (l *Single) Stop() { + close(l.stopChan) + + if l.cancel != nil { + l.cancel() + } + + l.wg.Wait() + l.logger.Infof(l.ctx, "timer single: stopped") +} + +// 获取任务数量 +func (l *Single) TaskCount() int { + count := 0 + l.workerList.Range(func(k, v interface{}) bool { + count++ + return true + }) + return count +} + +func (l *Single) MaxIndex() int64 { + return atomic.LoadInt64(&l.timerIndex) + 1 +} + +// 定时器主循环 +func (l *Single) timerLoop() { + defer l.wg.Done() + + ticker := time.NewTicker(100 * time.Millisecond) // 提高精度到100ms + defer ticker.Stop() + + for { + select { + case t := <-ticker.C: + l.nextTimeMux.RLock() + nextTime := l.nextTime + l.nextTimeMux.RUnlock() + + if t.Before(nextTime) { + continue + } + + l.iterator(l.ctx) + + case <-l.ctx.Done(): + l.logger.Infof(l.ctx, "timer: context cancelled, stopping timer loop") + return + case <-l.stopChan: + l.logger.Infof(l.ctx, "timer: received stop signal, stopping timer loop") + return + } + } +} + +// 清理循环 +func (s *Single) cleanupLoop() { + defer s.wg.Done() + + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + now := time.Now() + cleanupTime := now.Add(-2 * time.Minute) // 清理2分钟前的记录 + + s.hasRun.Range(func(k, v any) bool { + t, ok := v.(time.Time) + if !ok || t.Before(cleanupTime) { + s.hasRun.Delete(k) + } + return true + }) + + case <-s.ctx.Done(): + s.logger.Infof(s.ctx, "timer: context cancelled, stopping cleanup loop") + return + case <-s.stopChan: + s.logger.Infof(s.ctx, "timer: received stop signal, stopping cleanup loop") + return + } + } +} + +// 每月执行一次 +// @param ctx 上下文 +// @param taskId 任务ID +// @param day 每月的几号 +// @param hour 小时 +// @param minute 分钟 +// @param second 秒 +// @param callback 回调函数 +// @param extendData 扩展数据 +// @return error +func (c *Single) EveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { + + // nowTime := time.Now().In(c.location) + + jobData := JobData{ + JobType: JobTypeEveryMonth, + TaskId: taskId, + // CreateTime: nowTime, + Day: day, + Hour: hour, + Minute: minute, + Second: second, + } + + return c.addJob(ctx, jobData, callback, extendData) +} + +// 每周执行一次 +// @param ctx context.Context 上下文 +// @param taskId string 任务ID +// @param week time.Weekday 周 +// @param hour int 小时 +// @param minute int 分钟 +// @param second int 秒 +func (c *Single) EveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { + // nowTime := time.Now().In(c.location) + + jobData := JobData{ + JobType: JobTypeEveryWeek, + TaskId: taskId, + // CreateTime: nowTime, + Weekday: week, + Hour: hour, + Minute: minute, + Second: second, + } + + return c.addJob(ctx, jobData, callback, extendData) +} + +// 每天执行一次 +func (c *Single) EveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { + // nowTime := time.Now().In(c.location) + + jobData := JobData{ + JobType: JobTypeEveryDay, + TaskId: taskId, + // CreateTime: nowTime, + Hour: hour, + Minute: minute, + Second: second, + } + + return c.addJob(ctx, jobData, callback, extendData) +} + +// 每小时执行一次 +func (c *Single) EveryHour(ctx context.Context, taskId string, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { + // nowTime := time.Now().In(c.location) + + jobData := JobData{ + JobType: JobTypeEveryHour, + TaskId: taskId, + // CreateTime: nowTime, + Minute: minute, + Second: second, + } + + return c.addJob(ctx, jobData, callback, extendData) +} + +// 每分钟执行一次 +func (c *Single) EveryMinute(ctx context.Context, taskId string, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { + // nowTime := time.Now().In(c.location) + + jobData := JobData{ + JobType: JobTypeEveryMinute, + TaskId: taskId, + // CreateTime: nowTime, + Second: second, + } + + return c.addJob(ctx, jobData, callback, extendData) +} + +// 特定时间间隔 +func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) { + + if spaceTime < 0 { + c.logger.Errorf(ctx, "间隔时间不能小于0") + return 0, ErrIntervalTime + } + + // 固定时间点为20250101 00:00:00，便于计算下一次执行时间 + zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, c.location) + + jobData := JobData{ + JobType: JobTypeInterval, + TaskId: taskId, + // CreateTime: nowTime, + BaseTime: zeroTime, + IntervalTime: spaceTime, + } + + return c.addJob(ctx, jobData, callback, extendData) +} + +func (l *Single) Cron(ctx context.Context, taskId string, cronExpression string, callback func(ctx context.Context, extendData any) error, extendData any, opt ...Option) (int64, error) { + // 固定时间点为20250101 00:00:00，便于计算下一次执行时间 + zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, l.location) + + options := Options{} + for _, o := range opt { + o(&options) + } + cronParser := l.cronParser + if options.cronParser != nil { + cronParser = options.cronParser + } + + sche, err := GetCronSche(cronExpression, cronParser) + if err != nil { + l.logger.Errorf(ctx, "timer Single Cron cronExpression error:%s", err.Error()) + return 0, err + } + + jobData := JobData{ + JobType: JobTypeCron, + TaskId: taskId, + BaseTime: zeroTime, // 默认当天的零点 + CronExpression: cronExpression, + CronSchedule: sche, + } + return l.addJob(ctx, jobData, callback, extendData) +} + +// 间隔定时器 +// @param space 间隔时间 +// @param call 回调函数 +// @param extend 附加参数 +// @return int 定时器索引 +// @return error 错误 +func (l *Single) addJob(ctx context.Context, jobData JobData, call func(ctx context.Context, extendData interface{}) error, extend interface{}) (int64, error) { + if jobData.TaskId == "" { + l.logger.Errorf(ctx, "任务ID不能为空") + return 0, ErrTaskIdEmpty + } + if jobData.Day < 0 || jobData.Day > 31 { + l.logger.Errorf(ctx, "每月的天数必须在0-31之间") + return 0, ErrMonthDay + } + if jobData.Hour < 0 || jobData.Hour > 23 { + l.logger.Errorf(ctx, "小时必须在0-23之间") + return 0, ErrHour + } + if jobData.Minute < 0 || jobData.Minute > 59 { + l.logger.Errorf(ctx, "分钟必须在0-59之间") + return 0, ErrMinute + } + if jobData.Second < 0 || jobData.Second > 59 { + l.logger.Errorf(ctx, "秒必须在0-59之间") + return 0, ErrSecond + } + if call == nil { + l.logger.Errorf(ctx, "回调函数不能为空") + return 0, ErrCallbackEmpty + } + + nextTime, err := GetNextTime(time.Now().In(l.location), jobData) + if err != nil { + l.logger.Errorf(ctx, "获取下次执行时间失败:%s", err.Error()) + return 0, err + } + jobData.NextTime = *nextTime + + // 生成唯一索引 + index := atomic.AddInt64(&l.timerIndex, 1) + + t := timerStr{ + Callback: call, + CanRunning: make(chan struct{}, 1), + ExtendData: extend, + TaskId: jobData.TaskId, + JobData: &jobData, + } + + l.workerList.Store(index, t) + + // 计算下次执行时间(全局) + l.updateNextTimeIfEarlier(*nextTime) + + return index, nil +} + +// 如果更早则更新下次执行时间(全局) +func (s *Single) updateNextTimeIfEarlier(candidate time.Time) { + s.nextTimeMux.Lock() + defer s.nextTimeMux.Unlock() + + if candidate.Before(s.nextTime) { + s.nextTime = candidate + } +} + +// 删除定时器 +func (l *Single) Del(index int64) { + if _, ok := l.workerList.Load(index); ok { + l.workerList.Delete(index) + } +} + +func (l *Single) DelByTaskId(taskId string) { + l.workerList.Range(func(k, v interface{}) bool { + timeStr, ok := v.(timerStr) + if ok && timeStr.TaskId == taskId { + l.workerList.Delete(k) + } + return true + }) +} + +// 迭代定时器列表 +func (l *Single) iterator(ctx context.Context) { + // 当前时间 + nowTime := time.Now().In(l.location) + + // 默认5秒后（如果没有值就暂停进来5秒） + newNextTime := nowTime.Add(time.Second * 5) + + l.workerList.Range(func(k, v interface{}) bool { + timeStr, ok := v.(timerStr) + if !ok { + l.logger.Errorf(ctx, "timer: 类型断言失败，跳过该任务") + l.workerList.Delete(k) + return true + } + + if timeStr.JobData.NextTime.Before(nowTime) || timeStr.JobData.NextTime.Equal(nowTime) { + + originTime := timeStr.JobData.NextTime + + // 计算下次执行时间 + nextTime, err := GetNextTime(nowTime, *timeStr.JobData) + if err != nil { + l.logger.Errorf(ctx, "timer: 计算下次执行时间失败:%s", err.Error()) + return true + } + // 更新下次执行时间 + timeStr.JobData.NextTime = *nextTime + + if nextTime.Before(newNextTime) { + // 本规则下次发送时间小于系统下次需要执行的时间：替换 + newNextTime = *nextTime + } + + go l.executeTask(ctx, timeStr, originTime) + + } + + return true + + }) + + l.updateNextTime(newNextTime) +} + +// 执行任务 +func (s *Single) executeTask(ctx context.Context, timer timerStr, originTime time.Time) { + // 创建带追踪ID的上下文 + + u, err := uuid.NewV7() + if err != nil { + s.logger.Errorf(ctx, "Single executeTask uuid.NewV7 err:%v, using fallback", err) + u = uuid.New() + } + + traceCtx := context.WithValue(ctx, "trace_id", u.String()) + s.logger.Infof(traceCtx, "timer Single begin taskId:%s originTime:%d", timer.TaskId, originTime.UnixMilli()) + traceCtx, cancel := context.WithTimeout(traceCtx, s.timeout) // 设置执行超时 + defer cancel() + + select { + case timer.CanRunning <- struct{}{}: + defer func() { + select { + case <-timer.CanRunning: + default: + } + }() + + // 执行回调 + begin := time.Now() + if err := s.doTask(traceCtx, timer, originTime); err != nil { + s.logger.Errorf(traceCtx, "timer: 任务执行失败: %s", err.Error()) + } + s.logger.Infof(traceCtx, "timer Single end taskId:%s originTime:%d cost:%dms", timer.TaskId, originTime.UnixMilli(), time.Since(begin).Milliseconds()) + case <-traceCtx.Done(): + s.logger.Errorf(traceCtx, "timer: 任务执行超时: %s", timer.TaskId) + default: + // 任务正在执行中，跳过本次 + s.logger.Infof(traceCtx, "timer: 任务正在执行中，跳过本次 %s", timer.TaskId) + } +} + +// 定时器操作类 +// 这里不应painc +func (l *Single) doTask(ctx context.Context, timeStr timerStr, originTime time.Time) error { + // 检查任务是否已执行 + taskKey := fmt.Sprintf("%s:%d", timeStr.TaskId, originTime.UnixMilli()) + if _, loaded := l.hasRun.LoadOrStore(taskKey, time.Now()); loaded { + l.logger.Errorf(ctx, "timer: 任务已执行，跳过本次执行 %s", timeStr.TaskId) + return ErrTaskExecuted + } + + defer func() { + if err := recover(); err != nil { + l.logger.Errorf(ctx, "timer Single call panic err:%+v stack:%s", err, string(debug.Stack())) + } + }() + + err := timeStr.Callback(ctx, timeStr.ExtendData) + if err != nil { + l.logger.Errorf(ctx, "timer Single call back %s, err: %v", timeStr.TaskId, err) + return err + } + + return nil +} + +// 更新下次执行时间 +func (s *Single) updateNextTime(newTime time.Time) { + s.nextTimeMux.Lock() + defer s.nextTimeMux.Unlock() + + now := time.Now() + if newTime.Before(now) { + s.nextTime = now + } else { + s.nextTime = newTime + } +} diff --git a/single_test.go b/single_test.go index 064a18b..4933c8a 100644 --- a/single_test.go +++ b/single_test.go @@ -410,23 +410,19 @@ func TestSingleTimer_TaskDeletion(t *testing.T) { assert.Equal(t, 0, timer.TaskCount()) } -// 测试GetNextTime函数（需要根据实际实现调整） func TestGetNextTime2(t *testing.T) { now := time.Now().UTC() + baseTime := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), 0, 0, time.UTC) - // 测试间隔任务 jobData := timerx.JobData{ JobType: timerx.JobTypeInterval, IntervalTime: time.Minute, - // CreateTime: now, - BaseTime: now, + BaseTime: baseTime, } - tt := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), 0, 0, time.UTC) - nextTime, err := timerx.GetNextTime(now, jobData) assert.NoError(t, err) - assert.WithinDuration(t, tt.Add(time.Minute), *nextTime, time.Second) + assert.WithinDuration(t, baseTime.Add(time.Minute), *nextTime, time.Second) } // 基准测试