diff --git a/cluster.go b/cluster.go index db8c45c..09cb949 100644 --- a/cluster.go +++ b/cluster.go @@ -70,7 +70,11 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin ctx, cancel := context.WithCancel(ctx) - U, _ := uuid.NewV7() + U, err := uuid.NewV7() + if err != nil { + op.logger.Errorf(ctx, "InitCluster uuid.NewV7 err:%v, using fallback", err) + U = uuid.New() + } clu := &Cluster{ ctx: ctx, @@ -204,7 +208,7 @@ func (l *Cluster) startDaemon() { } func (l *Cluster) cleanExecuteInfoLoop() { - l.wg.Done() + defer l.wg.Done() ticker := time.NewTicker(time.Minute * 5) defer ticker.Stop() @@ -224,10 +228,11 @@ func (l *Cluster) cleanExecuteInfoLoop() { } // 清除过期任务 -func (l *Cluster) cleanExecuteInfo() error { +func (l *Cluster) cleanExecuteInfo() { // 移除执行信息 - l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err() - return nil + if err := l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err(); err != nil { + l.logger.Errorf(l.ctx, "cleanExecuteInfo ZRemRangeByScore err:%v", err) + } } // scheduleTasks 调度任务(只有leader执行) @@ -449,7 +454,11 @@ func (l *Cluster) calculateNextTimes() { // 根据内部注册的任务列表计算下一次执行的时间 l.workerList.Range(func(key, value interface{}) bool { - val := value.(timerStr) + val, ok := value.(timerStr) + if !ok { + l.logger.Errorf(l.ctx, "Cluster calculateNextTimes invalid type in workerList for key:%v", key) + return true + } nextTime, err := GetNextTime(time.Now().In(l.location), *val.JobData) if err != nil { @@ -539,7 +548,7 @@ func (c *Cluster) executeTasks() { case <-c.ctx.Done(): return case c.workerChan <- struct{}{}: - func() { + go func() { defer func() { <-c.workerChan }() @@ -553,7 +562,6 @@ func (c *Cluster) executeTasks() { if err != nil { if err != redis.Nil { c.logger.Errorf(c.ctx, "Failed to pop task: %v", err) - // Redis 异常,休眠一会儿 time.Sleep(5 * time.Second) } return @@ -561,22 +569,16 @@ func (c *Cluster) executeTasks() { if len(taskID) < 2 { c.logger.Errorf(c.ctx, "Invalid BLPop result: %v", taskID) - // 数据异常,继续下一个 return } - go c.processTask(taskID[1]) + c.processTask(taskID[1]) }() } } } -type ReJobData struct { - TaskId string - Times int -} - // 执行任务 func (l *Cluster) processTask(taskId string) { @@ -585,7 +587,11 @@ func (l *Cluster) processTask(taskId string) { ctx, cancel := context.WithTimeout(l.ctx, l.timeout) defer cancel() - u, _ := uuid.NewV7() + u, err := uuid.NewV7() + if err != nil { + l.logger.Errorf(ctx, "processTask uuid.NewV7 err:%v, using fallback", err) + u = uuid.New() + } ctx = context.WithValue(ctx, "trace_id", u.String()) @@ -593,10 +599,12 @@ func (l *Cluster) processTask(taskId string) { // 上报执行情况 executeVal := fmt.Sprintf("tid:%s|insId:%s|uuid:%s|time:%s", taskId, l.instanceId, u.String(), begin.Format(time.RFC3339Nano)) - l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{ + if err := l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{ Score: float64(begin.UnixMilli()), Member: executeVal, - }) + }).Err(); err != nil { + l.logger.Errorf(ctx, "processTask ZAdd executeInfo err:%v", err) + } val, ok := l.workerList.Load(taskId) if !ok { diff --git a/heartbeat/heartbeat.go b/heartbeat/heartbeat.go index e4330a1..e67e15b 100644 --- a/heartbeat/heartbeat.go +++ b/heartbeat/heartbeat.go @@ -150,7 +150,9 @@ func (l *HeartBeat) cleanHeartbeat(cleanSelf bool) error { } // 移除心跳 - l.redis.ZRemRangeByScore(l.ctx, l.heartbeatKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Second).UnixMilli(), 10)).Err() + if err := l.redis.ZRemRangeByScore(l.ctx, l.heartbeatKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Second).UnixMilli(), 10)).Err(); err != nil { + l.logger.Errorf(l.ctx, "cleanHeartbeat ZRemRangeByScore err:%v", err) + } return nil } diff --git a/heartbeat/options.go b/heartbeat/options.go index b2a8c7c..9c69038 100644 --- a/heartbeat/options.go +++ b/heartbeat/options.go @@ -17,7 +17,10 @@ type Options struct { func defaultOptions() Options { - u, _ := uuid.NewV7() + u, err := uuid.NewV7() + if err != nil { + u = uuid.New() + } return Options{ logger: logger.NewLogger(), diff --git a/leader/leader.go b/leader/leader.go index eea2e12..f9edbf2 100644 --- a/leader/leader.go +++ b/leader/leader.go @@ -111,8 +111,11 @@ func (l *Leader) getLeaderLock() error { l.logger.Errorf(l.ctx, "getLeaderLock err:%+v", err) return err } - if b, _ := lock.Lock(); !b { - // 加锁失败 非Reader + if b, err := lock.Lock(); !b || err != nil { + // 加锁失败 非Leader + if err != nil { + l.logger.Errorf(l.ctx, "getLeaderLock lock.Lock err:%+v", err) + } l.leaderLock.Lock() l.isLeader = false l.leaderLock.Unlock() @@ -152,10 +155,10 @@ func (l *Leader) getLeaderLock() error { // 等待超时退出 <-lock.GetCtx().Done() - // 已过期 - // l.leaderLock.Lock() - // l.isLeader = false - // l.leaderLock.Unlock() + // 锁已过期,重置leader状态 + l.leaderLock.Lock() + l.isLeader = false + l.leaderLock.Unlock() return nil diff --git a/leader/options.go b/leader/options.go index 01ea794..b4992b3 100644 --- a/leader/options.go +++ b/leader/options.go @@ -15,7 +15,10 @@ type Options struct { func defaultOptions() Options { - u, _ := uuid.NewV7() + u, err := uuid.NewV7() + if err != nil { + u = uuid.New() + } return Options{ logger: logger.NewLogger(), diff --git a/next_time.go b/next_time.go index 252207e..a5326bf 100644 --- a/next_time.go +++ b/next_time.go @@ -240,56 +240,18 @@ func calculateNextCronTime(t time.Time, job JobData) (*time.Time, error) { return nil, ErrCronParser } - s := *job.CronSchedule - - next := s.Next(t) + next := job.CronSchedule.Next(t) return &next, nil } -func GetCronSche(CronExpression string, cronParser *cron.Parser) (*cron.Schedule, error) { +func GetCronSche(CronExpression string, cronParser *cron.Parser) (cron.Schedule, error) { if CronExpression == "" { return nil, ErrCronExpression } if cronParser == nil { return nil, ErrCronParser } - sche, err := cronParser.Parse(CronExpression) - if err != nil { - return nil, err - } - return &sche, nil + return cronParser.Parse(CronExpression) } -// 检查是否本周期可以运行 -// 检查是否本周期可以运行(已弃用,使用新的时间比较逻辑) -// 保留此函数用于向后兼容,但建议使用新的时间计算逻辑 -func canRun(t time.Time, job JobData) bool { - targetTime := time.Date(t.Year(), t.Month(), t.Day(), job.Hour, job.Minute, job.Second, 0, t.Location()) - switch job.JobType { - case JobTypeEveryMonth: - // 对于月任务,需要比较日期 - targetTime = time.Date(t.Year(), t.Month(), job.Day, job.Hour, job.Minute, job.Second, 0, t.Location()) - return !targetTime.Before(t) - case JobTypeEveryWeek: - // 对于周任务,需要比较星期 - currentWeekday := t.Weekday() - if currentWeekday < job.Weekday { - return true - } - if currentWeekday == job.Weekday { - return targetTime.After(t) || targetTime.Equal(t) - } - return false - case JobTypeEveryDay: - return targetTime.After(t) || targetTime.Equal(t) - case JobTypeEveryHour: - hourTarget := time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), job.Minute, job.Second, 0, t.Location()) - return hourTarget.After(t) || hourTarget.Equal(t) - case JobTypeEveryMinute: - minuteTarget := time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), job.Second, 0, t.Location()) - return minuteTarget.After(t) || minuteTarget.Equal(t) - default: - return false - } -} diff --git a/once.go b/once.go index eee2a04..4c1e1c7 100644 --- a/once.go +++ b/once.go @@ -108,7 +108,11 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c ctx, cancel := context.WithCancel(ctx) - u, _ := uuid.NewV7() + u, err := uuid.NewV7() + if err != nil { + op.logger.Errorf(ctx, "InitOnce uuid.NewV7 err:%v, using fallback", err) + u = uuid.New() + } wo := &Once{ ctx: ctx, @@ -233,7 +237,7 @@ func (l *Once) startDaemon() { } func (l *Once) cleanExecuteInfoLoop() { - l.wg.Done() + defer l.wg.Done() ticker := time.NewTicker(time.Minute * 5) defer ticker.Stop() @@ -253,10 +257,11 @@ func (l *Once) cleanExecuteInfoLoop() { } // 清除过期任务 -func (l *Once) cleanExecuteInfo() error { +func (l *Once) cleanExecuteInfo() { // 移除执行信息 - l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err() - return nil + if err := l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err(); err != nil { + l.logger.Errorf(l.ctx, "cleanExecuteInfo ZRemRangeByScore err:%v", err) + } } // 任务调度 领导 @@ -302,7 +307,7 @@ func (l *Once) executeTasks() { case <-l.ctx.Done(): return case l.workerChan <- struct{}{}: - func() { + go func() { defer func() { <-l.workerChan }() @@ -316,7 +321,6 @@ func (l *Once) executeTasks() { if err != nil { if err != redis.Nil { l.logger.Errorf(l.ctx, "Failed to pop task: %v", err) - // Redis 异常,休眠一会儿再重试 time.Sleep(time.Second * 5) } return @@ -324,11 +328,10 @@ func (l *Once) executeTasks() { if len(keys) < 2 { l.logger.Errorf(l.ctx, "Invalid task data: %v", keys) - // 数据异常,继续下一个 return } // 处理任务 - go l.processTask(keys[1]) + l.processTask(keys[1]) }() } } @@ -409,7 +412,11 @@ func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType, RunCount: runCount, JobType: jobType, } - b, _ := json.Marshal(ed) + b, err := json.Marshal(ed) + if err != nil { + w.logger.Errorf(ctx, "save json.Marshal err:%v taskType:%v taskId:%v", err, taskType, taskId) + return err + } // 使用事务确保原子性 pipe := w.redis.TxPipeline() @@ -424,7 +431,7 @@ func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType, Score: float64(nextTime.UnixMilli()), Member: redisKey, }) - _, err := pipe.Exec(w.ctx) + _, err = pipe.Exec(w.ctx) if err != nil { w.logger.Errorf(w.ctx, "save task failed:%v taskType:%v taskId:%v attachData:%v retryCount:%v", err, taskType, taskId, attachData, runCount) return err @@ -509,8 +516,21 @@ func (w *Once) Delete(taskType OnceTaskType, taskId string) error { } // 获取任务 -func (l *Once) Get(taskType OnceTaskType, taskId string) { - // +func (l *Once) Get(taskType OnceTaskType, taskId string) (any, error) { + redisKey := l.buildRedisKey(taskType, taskId) + dataKey := fmt.Sprintf("timer:{%s}:once_data:%s", l.keyPrefix, redisKey) + str, err := l.redis.Get(l.ctx, dataKey).Result() + if err != nil { + if errors.Is(err, redis.Nil) { + return nil, nil + } + return nil, err + } + var ed extendData + if err := json.Unmarshal([]byte(str), &ed); err != nil { + return nil, err + } + return ed.Data, nil } // 批量获取任务 @@ -551,7 +571,11 @@ func (l *Once) processTask(key string) { ctx, cancel := context.WithTimeout(l.ctx, l.timeout) defer cancel() - u, _ := uuid.NewV7() + u, err := uuid.NewV7() + if err != nil { + l.logger.Errorf(ctx, "processTask uuid.NewV7 err:%v, using fallback", err) + u = uuid.New() + } ctx = context.WithValue(ctx, "trace_id", u.String()) @@ -577,10 +601,12 @@ func (l *Once) processTask(key string) { // 上报执行情况 executeVal := fmt.Sprintf("tid:%s|insId:%s|uuid:%s|time:%s", key, l.instanceId, u.String(), begin.Format(time.RFC3339Nano)) - l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{ + if err := l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{ Score: float64(begin.UnixMilli()), Member: executeVal, - }) + }).Err(); err != nil { + l.logger.Errorf(ctx, "processTask ZAdd executeInfo err:%v", err) + } defer func() { if err := recover(); err != nil { @@ -597,7 +623,10 @@ func (l *Once) processTask(key string) { } ed := extendData{} - json.Unmarshal([]byte(str), &ed) + if err := json.Unmarshal([]byte(str), &ed); err != nil { + l.logger.Errorf(ctx, "processTask json.Unmarshal err:%v key:%s", err, key) + return + } resp := l.worker.Worker(ctx, taskType, taskId, ed.Data) l.logger.Infof(ctx, "processTask exec key:%s resp:%+v data:%s", key, resp, str) diff --git a/option.go b/option.go index c96373c..9fd2aef 100644 --- a/option.go +++ b/option.go @@ -123,7 +123,7 @@ func WithMaxRetryCount(count int) Option { func WithMaxWorkers(count int) Option { return func(o *Options) { if count < 0 { - count = 10 + count = 100 } o.maxWorkers = count } diff --git a/priority/option.go b/priority/option.go index bef428f..01b0835 100644 --- a/priority/option.go +++ b/priority/option.go @@ -18,7 +18,10 @@ type Options struct { func defaultOptions() Options { - u, _ := uuid.NewV7() + u, err := uuid.NewV7() + if err != nil { + u = uuid.New() + } return Options{ getInterval: time.Second * 2, @@ -47,7 +50,10 @@ func WithLogger(log logger.Logger) Option { // 更新周期 func WithUpdateInterval(d time.Duration) Option { - if d.Abs() < time.Second { + if d <= 0 { + d = time.Second * 5 + } + if d < time.Second { d = time.Second * 5 } return func(o *Options) { diff --git a/priority/priority.go b/priority/priority.go index 80f360d..e4d82ff 100644 --- a/priority/priority.go +++ b/priority/priority.go @@ -182,7 +182,10 @@ func (p *Priority) setPriority() (string, error) { // 解析结果 if resultMap, ok := result.([]interface{}); ok && len(resultMap) == 1 { - resultStr := resultMap[0].(string) + resultStr, ok := resultMap[0].(string) + if !ok { + return "", fmt.Errorf("script unexpected result type: %T", resultMap[0]) + } return resultStr, nil } diff --git a/types.go b/types.go index 4c6d77e..9816e5b 100644 --- a/types.go +++ b/types.go @@ -41,7 +41,7 @@ type JobData struct { Minute int // 每小时的第几分钟 Second int // 每分钟的第几秒 CronExpression string // cron表达式 - CronSchedule *cron.Schedule // cron表达式解析后的数据 + CronSchedule cron.Schedule // cron表达式解析后的数据 } // 定义各个回调函数