优化部分细节

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
 	}

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)

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,
 		},
 		{

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
 	}

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

single.go

@@ -435,7 +435,11 @@ func (l *Single) iterator(ctx context.Context) {
 func (s *Single) executeTask(ctx context.Context, timer timerStr, originTime time.Time) {
 	// 创建带追踪ID的上下文
 
-	u, _ := uuid.NewV7()
+	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())

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:     baseTime,
-		BaseTime: now,
 	}
 
-	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)
 }
 
 // 基准测试