添加一些描述

cluster.go

@@ -9,8 +9,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/redis/go-redis/v9"
 	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
+	"github.com/robfig/cron/v3"
 	"github.com/yuninks/cachex"
 	"github.com/yuninks/lockx"
 	"github.com/yuninks/timerx/heartbeat"
@@ -51,6 +52,10 @@ type Cluster struct {
 	leader     *leader.Leader       // Leader
 	heartbeat  *heartbeat.HeartBeat // 心跳
 	cache      *cachex.Cache        // 本地缓存
+	cronParser *cron.Parser         // cron表达式解析器
+	batchSize  int                  // 批量获取任务的数量
+	workerChan chan struct{}        // worker
+	maxWorkers int                  // 最大worker数量
 }
 
 // 初始化定时器
@@ -82,19 +87,34 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 		setKey:         "timer:cluster_setKey" + keyPrefix,         // 重入集合
 		priorityKey:    "timer:cluster_priorityKey" + keyPrefix,    // 全局优先级的key
 		executeInfoKey: "timer:cluster_executeInfoKey" + keyPrefix, // 执行情况的key 有序集合
-		usePriority:    op.usePriority,
+		usePriority:    false,
 		stopChan:       make(chan struct{}),
 		instanceId:     U.String(),
+		cronParser:     op.cronParser,
+		batchSize:      op.batchSize,
+		workerChan:     make(chan struct{}, op.maxWorkers),
+		maxWorkers:     op.maxWorkers,
 	}
 
 	// 初始化优先级
 
-	if clu.usePriority {
+	if op.priorityType != priorityTypeNone {
+
+		clu.usePriority = true
+
+		if op.priorityType == priorityTypeVersion {
+			pVal, err := priority.PriorityByVersion(op.priorityVersion)
+			if err != nil {
+				clu.logger.Errorf(ctx, "PriorityByVersion version:%s err:%v", op.priorityVersion, err)
+				return nil, err
+			}
+			op.priorityVal = pVal
+		}
 
 		pri, err := priority.InitPriority(
 			ctx,
 			red,
-			clu.priorityKey,
+			clu.keyPrefix,
 			op.priorityVal,
 			priority.WithLogger(clu.logger),
 			priority.WithInstanceId(clu.instanceId),
@@ -111,7 +131,7 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 	le, err := leader.InitLeader(
 		ctx,
 		clu.redis,
-		keyPrefix,
+		clu.keyPrefix,
 		leader.WithLogger(clu.logger),
 		leader.WithPriority(clu.priority),
 		leader.WithInstanceId(clu.instanceId),
@@ -132,7 +152,7 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 		heartbeat.WithLeader(clu.leader),
 		heartbeat.WithLogger(clu.logger),
 		heartbeat.WithPriority(clu.priority),
-		heartbeat.WithSource("once"),
+		heartbeat.WithSource("cluster"),
 	)
 	if err != nil {
 		clu.logger.Errorf(ctx, "InitHeartBeat err:%v", err)
@@ -251,6 +271,7 @@ func (c *Cluster) EveryMonth(ctx context.Context, taskId string, day int, hour i
 
 	jobData := JobData{
 		JobType: JobTypeEveryMonth,
+		TaskId:  taskId,
 		// CreateTime: nowTime,
 		Day:    day,
 		Hour:   hour,
@@ -258,7 +279,7 @@ func (c *Cluster) EveryMonth(ctx context.Context, taskId string, day int, hour i
 		Second: second,
 	}
 
-	return c.addJob(ctx, taskId, jobData, callback, extendData)
+	return c.addJob(ctx, jobData, callback, extendData)
 }
 
 // 每周执行一次
@@ -273,6 +294,7 @@ func (c *Cluster) EveryWeek(ctx context.Context, taskId string, week time.Weekda
 
 	jobData := JobData{
 		JobType: JobTypeEveryWeek,
+		TaskId:  taskId,
 		// CreateTime: nowTime,
 		Weekday: week,
 		Hour:    hour,
@@ -280,7 +302,7 @@ func (c *Cluster) EveryWeek(ctx context.Context, taskId string, week time.Weekda
 		Second:  second,
 	}
 
-	return c.addJob(ctx, taskId, jobData, callback, extendData)
+	return c.addJob(ctx, jobData, callback, extendData)
 }
 
 // 每天执行一次
@@ -289,13 +311,14 @@ func (c *Cluster) EveryDay(ctx context.Context, taskId string, hour int, minute
 
 	jobData := JobData{
 		JobType: JobTypeEveryDay,
+		TaskId:  taskId,
 		// CreateTime: nowTime,
 		Hour:   hour,
 		Minute: minute,
 		Second: second,
 	}
 
-	return c.addJob(ctx, taskId, jobData, callback, extendData)
+	return c.addJob(ctx, jobData, callback, extendData)
 }
 
 // 每小时执行一次
@@ -304,12 +327,13 @@ func (c *Cluster) EveryHour(ctx context.Context, taskId string, minute int, seco
 
 	jobData := JobData{
 		JobType: JobTypeEveryHour,
+		TaskId:  taskId,
 		// CreateTime: nowTime,
 		Minute: minute,
 		Second: second,
 	}
 
-	return c.addJob(ctx, taskId, jobData, callback, extendData)
+	return c.addJob(ctx, jobData, callback, extendData)
 }
 
 // 每分钟执行一次
@@ -318,32 +342,68 @@ func (c *Cluster) EveryMinute(ctx context.Context, taskId string, second int, ca
 
 	jobData := JobData{
 		JobType: JobTypeEveryMinute,
+		TaskId:  taskId,
 		// CreateTime: nowTime,
 		Second: second,
 	}
 
-	return c.addJob(ctx, taskId, jobData, callback, extendData)
+	return c.addJob(ctx, jobData, callback, extendData)
 }
 
 // 特定时间间隔
 func (c *Cluster) EverySpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
-	nowTime := time.Now().In(c.location)
 
 	if spaceTime < 0 {
 		c.logger.Errorf(ctx, "间隔时间不能小于0")
 		return errors.New("间隔时间不能小于0")
 	}
 
-	// 获取当天的零点时间
-	zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, c.location)
 
 	jobData := JobData{
 		JobType:      JobTypeInterval,
+		TaskId:       taskId,
 		BaseTime:     zeroTime, // 默认当天的零点
 		IntervalTime: spaceTime,
 	}
 
-	return c.addJob(ctx, taskId, jobData, callback, extendData)
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
+// 定时任务
+// 使用的是秒级cron表达式，可以使用Option设置cronParser
+// @param ctx context.Context 上下文
+// @param taskId string 任务ID
+// @param cronExpression string cron表达式
+// @param callback callback 回调函数
+// @param extendData interface{} 扩展数据
+// @return error
+func (l *Cluster) Cron(ctx context.Context, taskId string, cronExpression string, callback func(ctx context.Context, extendData any) error, extendData any, opt ...Option) error {
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, l.location)
+
+	options := newEmptyOptions(opt...)
+	cronParser := l.cronParser
+	if options.cronParser != nil {
+		cronParser = options.cronParser
+	}
+
+	sche, err := GetCronSche(cronExpression, cronParser)
+	if err != nil {
+		l.logger.Errorf(ctx, "Cron cronExpression error:%s", err.Error())
+		return err
+	}
+
+	jobData := JobData{
+		JobType:        JobTypeCron,
+		TaskId:         taskId,
+		BaseTime:       zeroTime, // 默认当天的零点
+		CronExpression: cronExpression,
+		CronSchedule:   sche,
+	}
+
+	return l.addJob(ctx, jobData, callback, extendData)
 }
 
 // 统一添加任务
@@ -353,11 +413,11 @@ func (c *Cluster) EverySpace(ctx context.Context, taskId string, spaceTime time.
 // @param callback callback 回调函数
 // @param extendData interface{} 扩展数据
 // @return error
-func (l *Cluster) addJob(ctx context.Context, taskId string, jobData JobData, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+func (l *Cluster) addJob(ctx context.Context, jobData JobData, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
 	// 判断是否重复
-	_, ok := l.workerList.Load(taskId)
+	_, ok := l.workerList.Load(jobData.TaskId)
 	if ok {
-		l.logger.Errorf(ctx, "Cluster addJob taskId exits:%s", taskId)
+		l.logger.Errorf(ctx, "Cluster addJob taskId exits:%s", jobData.TaskId)
 		return ErrTaskIdExists
 	}
 
@@ -371,13 +431,13 @@ func (l *Cluster) addJob(ctx context.Context, taskId string, jobData JobData, ca
 	t := timerStr{
 		Callback:   callback,
 		ExtendData: extendData,
-		TaskId:     taskId,
+		TaskId:     jobData.TaskId,
 		JobData:    &jobData,
 	}
 
-	l.workerList.Store(taskId, t)
+	l.workerList.Store(jobData.TaskId, t)
 
-	l.logger.Infof(ctx, "Cluster addJob taskId:%s", taskId)
+	l.logger.Infof(ctx, "Cluster addJob taskId:%s", jobData.TaskId)
 
 	return nil
 }
@@ -402,10 +462,10 @@ func (l *Cluster) calculateNextTimes() {
 		// 使用Lua脚本原子性添加任务
 		script := `
 			local zsetKey = KEYS[1]
-			local lockKey = KEYS[2]
 			local score = ARGV[1]
 			local taskID = ARGV[2]
 			local expireTime = ARGV[3]
+			local lockKey = ARGV[4]
 
 			-- 检查是否已存在
 			local existing = redis.call('zscore', zsetKey, taskID)
@@ -424,8 +484,8 @@ func (l *Cluster) calculateNextTimes() {
 			`
 
 		lockKey := fmt.Sprintf("%s_%s_%d", l.keyPrefix, val.TaskId, nextTime.UnixMilli())
-		_, err = pipe.Eval(l.ctx, script, []string{l.zsetKey, lockKey},
-			nextTime.UnixMilli(), val.TaskId, 60).Result()
+		_, err = pipe.Eval(l.ctx, script, []string{l.zsetKey},
+			nextTime.UnixMilli(), val.TaskId, 60, lockKey).Result()
 		if err != nil {
 			l.logger.Errorf(l.ctx, "Failed to schedule task: %v", err)
 		}
@@ -456,7 +516,7 @@ func (c *Cluster) moveReadyTasks() {
 	`
 
 	result, err := c.redis.Eval(c.ctx, script, []string{c.zsetKey, c.listKey},
-		time.Now().UnixMilli(), 100).Result()
+		time.Now().UnixMilli(), c.batchSize).Result()
 	if err != nil && err != redis.Nil {
 		c.logger.Errorf(c.ctx, "Failed to move ready tasks: %v", err)
 		return
@@ -472,15 +532,21 @@ func (c *Cluster) executeTasks() {
 	defer c.wg.Done()
 
 	for {
+
 		select {
 		case <-c.stopChan:
 			return
 		case <-c.ctx.Done():
 			return
-		default:
+		case c.workerChan <- struct{}{}:
+			func() {
+				defer func() {
+					<-c.workerChan
+				}()
+
 				if c.usePriority && !c.priority.IsLatest(c.ctx) {
 					time.Sleep(5 * time.Second)
-				continue
+					return
 				}
 
 				taskID, err := c.redis.BLPop(c.ctx, 10*time.Second, c.listKey).Result()
@@ -490,14 +556,17 @@ func (c *Cluster) executeTasks() {
 						// Redis 异常，休眠一会儿
 						time.Sleep(5 * time.Second)
 					}
-				continue
+					return
 				}
 
 				if len(taskID) < 2 {
-				continue
+					c.logger.Errorf(c.ctx, "Invalid BLPop result: %v", taskID)
+					// 数据异常，继续下一个
+					return
 				}
 
 				go c.processTask(taskID[1])
+			}()
 		}
 	}
 
@@ -510,6 +579,7 @@ type ReJobData struct {
 
 // 执行任务
 func (l *Cluster) processTask(taskId string) {
+
 	begin := time.Now()
 
 	ctx, cancel := context.WithTimeout(l.ctx, l.timeout)

cmd/main.go

@@ -7,6 +7,7 @@ import (
 	"time"
 
 	"github.com/redis/go-redis/v9"
+	"github.com/robfig/cron/v3"
 	"github.com/yuninks/timerx"
 	"github.com/yuninks/timerx/priority"
 )
@@ -26,14 +27,30 @@ func main() {
 
 	// re()
 	// d()
-	cluster()
+	// cluster()
 	// once()
+	single()
 	// prioritys()
 
 	select {}
 
 }
 
+func single() error {
+
+	ctx := context.Background()
+
+	ops := []timerx.Option{
+		timerx.WithCronParserSecond(),
+	}
+
+	single := timerx.InitSingle(ctx, ops...)
+
+	single.Cron(ctx, "test_cron1", "*/5 * * * * ?", callback, "这是cron任务1", timerx.WithCronParserSecond())
+
+	return nil
+}
+
 func prioritys() {
 
 	client := getRedis()
@@ -101,7 +118,7 @@ type OnceWorker struct{}
 
 func (l OnceWorker) Worker(ctx context.Context, taskType timerx.OnceTaskType, taskId string, attachData interface{}) *timerx.OnceWorkerResp {
 	// 追加写入文件
-	file, err := os.OpenFile("./test.txt", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	file, err := os.OpenFile("./test3.txt", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
 	if err != nil {
 		panic(err)
 	}
@@ -142,30 +159,45 @@ func cluster() {
 	// log := loggerx.NewLogger(ctx,loggerx.SetToConsole(),loggerx.SetEscapeHTML(false))
 	// _ = log
 
-	cluster, _ := timerx.InitCluster(ctx, client, "test", timerx.WithPriority(103))
-	err := cluster.EverySpace(ctx, "test_space1", 1*time.Second, aa, "这是秒任务1")
+	cluster, _ := timerx.InitCluster(ctx, client, "test2", timerx.WithPriority(104))
+	err := cluster.EverySpace(ctx, "test_space1", 1*time.Second, callback, "这是秒任务1")
 	fmt.Println(err)
-	err = cluster.EverySpace(ctx, "test_space2", 2*time.Second, aa, "这是秒任务2")
+	err = cluster.EverySpace(ctx, "test_space2", 2*time.Second, callback, "这是秒任务2")
 	fmt.Println(err)
-	err = cluster.EverySpace(ctx, "test_space3", 5*time.Second, aa, "这是秒任务3")
+	err = cluster.EverySpace(ctx, "test_space3", 5*time.Second, callback, "这是秒任务3")
 	fmt.Println(err)
 
-	err = cluster.EveryMinute(ctx, "test_min1", 15, aa, "这是分钟任务1")
+	err = cluster.EveryMinute(ctx, "test_min1", 15, callback, "这是分钟任务1")
 	fmt.Println(err)
-	err = cluster.EveryMinute(ctx, "test_min2", 30, aa, "这是分钟任务2")
+	err = cluster.EveryMinute(ctx, "test_min2", 30, callback, "这是分钟任务2")
 	fmt.Println(err)
 
-	err = cluster.EveryHour(ctx, "test_hour1", 30, 0, aa, "这是小时任务1")
+	err = cluster.EveryHour(ctx, "test_hour1", 30, 0, callback, "这是小时任务1")
 	fmt.Println(err)
-	err = cluster.EveryHour(ctx, "test_hour2", 30, 15, aa, "这是小时任务2")
+	err = cluster.EveryHour(ctx, "test_hour2", 30, 15, callback, "这是小时任务2")
 	fmt.Println(err)
 
-	err = cluster.EveryDay(ctx, "test_day1", 5, 0, 0, aa, "这是天任务1")
+	err = cluster.EveryDay(ctx, "test_day1", 5, 0, 0, callback, "这是天任务1")
 	fmt.Println(err)
-	err = cluster.EveryDay(ctx, "test_day2", 9, 20, 0, aa, "这是天任务2")
+	err = cluster.EveryDay(ctx, "test_day2", 9, 20, 0, callback, "这是天任务2")
 	fmt.Println(err)
-	err = cluster.EveryDay(ctx, "test_day3", 10, 30, 30, aa, "这是天任务3")
+	err = cluster.EveryDay(ctx, "test_day3", 10, 30, 30, callback, "这是天任务3")
 	fmt.Println(err)
+
+	// 默认秒级表达式
+	err = cluster.Cron(ctx, "test_cron1", "*/5 * * * * ?", callback, "这是cron任务1", timerx.WithCronParserSecond())
+	fmt.Println(err)
+	err = cluster.Cron(ctx, "test_cron2", "0/5 * * * * ?", callback, "这是cron任务2", timerx.WithCronParserSecond())
+	fmt.Println("这是cron任务2:", err)
+	// 自定义解析器
+	err = cluster.Cron(ctx, "test_cron3", "@every 2s", callback, "这是cron任务3", timerx.WithCronParserOption(cron.Descriptor))
+	fmt.Println("这是cron任务3:", err)
+	// Linux标准解析器
+	err = cluster.Cron(ctx, "test_cron4", "*/5 * * * *", callback, "这是cron任务4", timerx.WithCronParserLinux())
+	fmt.Println("这是cron任务4:", err)
+	// 仅符号解析器
+	err = cluster.Cron(ctx, "test_cron5", "@every 5s", callback, "这是cron任务5", timerx.WithCronParserDescriptor())
+	fmt.Println("这是cron任务5:", err)
 }
 
 func worker() {
@@ -208,17 +240,17 @@ func re() {
 
 	ctx := context.Background()
 	cl, _ := timerx.InitCluster(ctx, client, "kkkk")
-	cl.EverySpace(ctx, "test1", 1*time.Millisecond, aa, "data")
-	cl.EverySpace(ctx, "test2", 1*time.Millisecond, aa, "data")
-	cl.EverySpace(ctx, "test3", 1*time.Millisecond, aa, "data")
-	cl.EverySpace(ctx, "test4", 1*time.Millisecond, aa, "data")
-	cl.EverySpace(ctx, "test5", 1*time.Millisecond, aa, "data")
-	cl.EverySpace(ctx, "test6", 1*time.Millisecond, aa, "data")
+	cl.EverySpace(ctx, "test1", 1*time.Millisecond, callback, "data")
+	cl.EverySpace(ctx, "test2", 1*time.Millisecond, callback, "data")
+	cl.EverySpace(ctx, "test3", 1*time.Millisecond, callback, "data")
+	cl.EverySpace(ctx, "test4", 1*time.Millisecond, callback, "data")
+	cl.EverySpace(ctx, "test5", 1*time.Millisecond, callback, "data")
+	cl.EverySpace(ctx, "test6", 1*time.Millisecond, callback, "data")
 
 	select {}
 }
 
-func aa(ctx context.Context, data interface{}) error {
+func callback(ctx context.Context, data interface{}) error {
 
 	fmt.Println("-执行时间:", data, time.Now().Format("2006-01-02 15:04:05"))
 	// fmt.Println(data)

error.go

@@ -29,4 +29,16 @@ var (
 	ErrTaskIdExists = errors.New("taskId already exists")
 	// 任务已执行
 	ErrTaskExecuted = errors.New("task already executed")
+	// cron表达式错误
+	ErrCronExpression = errors.New("cron expression error")
+	// ErrCronParser 错误
+	ErrCronParser = errors.New("cron parser error")
+	// ExecuteTime 错误
+	ErrExecuteTime = errors.New("execute time error")
+	// RunCount 错误
+	ErrRunCount = errors.New("run count error")
+	// DelayTime 错误
+	ErrDelayTime = errors.New("delay time error")
+	// 任务已存在
+	ErrTaskExists = errors.New("task already exists")
 )

example/cluster/main.go

@@ -0,0 +1,299 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/yuninks/timerx"
+)
+
+var save_path = "./cluster.log"
+
+func main() {
+
+	ctx := context.Background()
+	client := getRedis()
+
+	ops := []timerx.Option{}
+
+	clu, err := timerx.InitCluster(ctx, client, "cluster_01", ops...)
+	if err != nil {
+		panic(err)
+	}
+
+	space(ctx, clu)
+	minute(ctx, clu)
+	hour(ctx, clu)
+	day(ctx, clu)
+	week(ctx, clu)
+	month(ctx, clu)
+	cron(ctx, clu)
+
+	select {}
+
+}
+
+// space
+func space(ctx context.Context, clu *timerx.Cluster) {
+	// 每秒执行一次
+	err := clu.EverySpace(ctx, "space_test_second_1", 1*time.Second, callback, "space 这是秒任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_second_5", 5*time.Second, callback, "space 这是5秒任务")
+	fmt.Println(err)
+
+	// 每分钟执行一次
+	err = clu.EverySpace(ctx, "space_test_minute_1", 1*time.Minute, callback, "space 这是分钟任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_minute_5", 5*time.Minute, callback, "space 这是5分钟任务")
+	fmt.Println(err)
+
+	// 每小时执行一次
+	err = clu.EverySpace(ctx, "space_test_hour_1", 1*time.Hour, callback, "space 这是小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_2", 2*time.Hour, callback, "space 这是2小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_3", 3*time.Hour, callback, "space 这是3小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_4", 4*time.Hour, callback, "space 这是4小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_5", 5*time.Hour, callback, "space 这是5小时任务")
+	fmt.Println(err)
+
+	// 每天执行一次
+	err = clu.EverySpace(ctx, "space_test_day_1", 24*time.Hour, callback, "space 这是天任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_day_2", 2*24*time.Hour, callback, "space 这是2天任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_day_3", 3*24*time.Hour, callback, "space 这是3天任务")
+	fmt.Println(err)
+
+	// 每周执行一次
+	err = clu.EverySpace(ctx, "space_test_week_1", 7*24*time.Hour, callback, "space 这是7天任务")
+	fmt.Println(err)
+
+	// 每月执行一次
+	err = clu.EverySpace(ctx, "space_test_month_1", 30*24*time.Hour, callback, "space 这是30天任务")
+	fmt.Println(err)
+
+}
+
+// minute
+func minute(ctx context.Context, clu *timerx.Cluster) {
+	// 每分钟0s执行一次
+	err := clu.EveryMinute(ctx, "minute_test_min1", 0, callback, "minute 这是分钟任务0")
+	fmt.Println(err)
+	// 每分钟5s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min5", 5, callback, "minute 这是分钟任务5")
+	fmt.Println(err)
+	// 每分钟10s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min10", 10, callback, "minute 这是分钟任务10")
+	fmt.Println(err)
+	// 每分钟15s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min15", 15, callback, "minute 这是分钟任务15")
+	fmt.Println(err)
+	// 每分钟30s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min30", 30, callback, "minute 这是分钟任务30")
+	fmt.Println(err)
+	// 每分钟45s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min45", 45, callback, "minute 这是分钟任务45")
+	fmt.Println(err)
+	// 每分钟50s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min50", 50, callback, "minute 这是分钟任务50")
+	fmt.Println(err)
+	// 每分钟55s执行一次
+	err = clu.EveryMinute(ctx, "minute_test_min55", 55, callback, "minute 这是分钟任务55")
+	fmt.Println(err)
+}
+
+// Hour
+func hour(ctx context.Context, clu *timerx.Cluster) {
+	// 每小时的第0分钟15s执行一次
+	err := clu.EveryHour(ctx, "hour_test_hour1", 0, 15, callback, "hour 这是小时任务1")
+	fmt.Println(err)
+	// 每小时的第5分钟30s执行一次
+	err = clu.EveryHour(ctx, "hour_test_hour2", 5, 30, callback, "hour 这是小时任务2")
+	fmt.Println(err)
+	// 每小时的第10分钟45s执行一次
+	err = clu.EveryHour(ctx, "hour_test_hour3", 10, 45, callback, "hour 这是小时任务3")
+	fmt.Println(err)
+	// 每小时的第15分钟0s执行一次
+	err = clu.EveryHour(ctx, "hour_test_hour4", 15, 0, callback, "hour 这是小时任务4")
+	fmt.Println(err)
+	// 每小时的第20分钟15s执行一次
+	err = clu.EveryHour(ctx, "hour_test_hour5", 20, 15, callback, "hour 这是小时任务5")
+	fmt.Println(err)
+}
+
+// Day
+func day(ctx context.Context, clu *timerx.Cluster) {
+	// 每天的00:00:00执行一次
+	err := clu.EveryDay(ctx, "day_test_day1", 0, 0, 0, callback, "day 这是天任务1")
+	fmt.Println(err)
+	// 每天的02:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day2", 2, 0, 0, callback, "day 这是天任务2")
+	fmt.Println(err)
+	// 每天的04:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day3", 4, 0, 0, callback, "day 这是天任务3")
+	fmt.Println(err)
+	// 每天的06:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day4", 6, 0, 0, callback, "day 这是天任务4")
+	fmt.Println(err)
+	// 每天的08:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day5", 8, 0, 0, callback, "day 这是天任务5")
+	fmt.Println(err)
+	// 每天的10:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day6", 10, 0, 0, callback, "day 这是天任务6")
+	fmt.Println(err)
+	// 每天的12:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day7", 12, 0, 0, callback, "day 这是天任务7")
+	fmt.Println(err)
+	// 每天的14:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day8", 14, 0, 0, callback, "day 这是天任务8")
+	fmt.Println(err)
+	// 每天的16:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day9", 16, 0, 0, callback, "day 这是天任务9")
+	fmt.Println(err)
+	// 每天的18:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day10", 18, 0, 0, callback, "day 这是天任务10")
+	fmt.Println(err)
+	// 每天的20:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day11", 20, 0, 0, callback, "day 这是天任务11")
+	fmt.Println(err)
+	// 每天的22:00:00执行一次
+	err = clu.EveryDay(ctx, "day_test_day12", 22, 0, 0, callback, "day 这是天任务12")
+	fmt.Println(err)
+}
+
+// Week
+func week(ctx context.Context, clu *timerx.Cluster) {
+	// 每周一 10:00:00 执行
+	err := clu.EveryWeek(ctx, "week_test_week1", 1, 10, 0, 0, callback, "week 这是周任务1")
+	fmt.Println(err)
+	// 每周二 10:00:00 执行
+	err = clu.EveryWeek(ctx, "week_test_week2", 2, 10, 0, 0, callback, "week 这是周任务2")
+	fmt.Println(err)
+	// 每周三 10:00:00 执行
+	err = clu.EveryWeek(ctx, "week_test_week3", 3, 10, 0, 0, callback, "week 这是周任务3")
+	fmt.Println(err)
+	// 每周四 10:00:00 执行
+	err = clu.EveryWeek(ctx, "week_test_week4", 4, 10, 0, 0, callback, "week 这是周任务4")
+	fmt.Println(err)
+	// 每周五 10:00:00 执行
+	err = clu.EveryWeek(ctx, "week_test_week5", 5, 10, 0, 0, callback, "week 这是周任务5")
+	fmt.Println(err)
+	// 每周六 10:00:00 执行
+	err = clu.EveryWeek(ctx, "week_test_week6", 6, 10, 0, 0, callback, "week 这是周任务6")
+	fmt.Println(err)
+	// 每周日 10:00:00 执行
+	err = clu.EveryWeek(ctx, "week_test_week7", 0, 10, 0, 0, callback, "week 这是周任务7")
+	fmt.Println(err)
+}
+
+// Month
+func month(ctx context.Context, clu *timerx.Cluster) {
+	// 每月的第1号 10:00:00 执行
+	err := clu.EveryMonth(ctx, "month_test_month1", 1, 10, 0, 0, callback, "month 这是月任务1")
+	fmt.Println(err)
+	// 每月的第5号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month5", 5, 10, 0, 0, callback, "month 这是月任务5")
+	fmt.Println(err)
+	// 每月的第10号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month10", 10, 10, 0, 0, callback, "month 这是月任务10")
+	fmt.Println(err)
+	// 每月的第15号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month15", 15, 10, 0, 0, callback, "month 这是月任务15")
+	fmt.Println(err)
+	// 每月的第20号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month20", 20, 10, 0, 0, callback, "month 这是月任务20")
+	fmt.Println(err)
+	// 每月的第25号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month25", 25, 10, 0, 0, callback, "month 这是月任务25")
+	fmt.Println(err)
+	// 每月的第28号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month28", 28, 10, 0, 0, callback, "month 这是月任务28")
+	fmt.Println(err)
+	// 每月的第29号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month29", 29, 10, 0, 0, callback, "month 这是月任务29")
+	fmt.Println(err)
+	// 每月的第30号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month30", 30, 10, 0, 0, callback, "month 这是月任务30")
+	fmt.Println(err)
+	// 每月的第31号 10:00:00 执行
+	err = clu.EveryMonth(ctx, "month_test_month31", 31, 10, 0, 0, callback, "month 这是月任务31")
+	fmt.Println(err)
+
+}
+
+func cron(ctx context.Context, clu *timerx.Cluster) {
+	// 秒级表达式 5秒执行一次
+	err := clu.Cron(ctx, "cron_test_cron1", "*/5 * * * * ?", callback, "cron 这是cron任务1", timerx.WithCronParserSecond())
+	fmt.Println(err)
+	// Linux表达式 5分钟执行一次
+	err = clu.Cron(ctx, "cron_test_cron2", "*/5 * * * *", callback, "cron 这是cron任务2", timerx.WithCronParserLinux())
+	fmt.Println(err)
+	// 符号表达式 5秒执行一次
+	err = clu.Cron(ctx, "cron_test_cron3", "@every 5s", callback, "cron 这是cron任务3", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每天执行一次
+	err = clu.Cron(ctx, "cron_test_cron4", "@daily", callback, "cron 这是cron任务4", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每月执行一次
+	err = clu.Cron(ctx, "cron_test_cron5", "@monthly", callback, "cron 这是cron任务5", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每年执行一次
+	err = clu.Cron(ctx, "cron_test_cron6", "@yearly", callback, "cron 这是cron任务6", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每周执行一次
+	err = clu.Cron(ctx, "cron_test_cron7", "@weekly", callback, "cron 这是cron任务7", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每小时执行一次
+	err = clu.Cron(ctx, "cron_test_cron8", "@hourly", callback, "cron 这是cron任务8", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每分钟执行一次
+	err = clu.Cron(ctx, "cron_test_cron9", "@minutely", callback, "cron 这是cron任务9", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+
+}
+
+func getRedis() *redis.Client {
+	client := redis.NewClient(&redis.Options{
+		Addr:     "127.0.0.1" + ":" + "6379",
+		Password: "123456", // no password set
+		DB:       0,        // use default DB
+	})
+	if client == nil {
+		panic("redis init error")
+	}
+	return client
+}
+
+func callback(ctx context.Context, extendData any) error {
+
+	fmt.Println("任务执行了", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
+	// 解析文件路径，每天一个文件
+	path, _ := filepath.Abs("./")
+	// 拼接文件路径
+	save_path = filepath.Join(path,"/cache/cluster/"+time.Now().Format("2006-01-02")+".log")
+	// 创建文件夹
+	dir := filepath.Dir(save_path)
+	os.MkdirAll(dir, 0755)
+
+	// 追加到文件
+	file, err := os.OpenFile(save_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		fmt.Println("打开文件失败:", err)
+		return err
+	}
+	defer file.Close()
+	_, err = file.WriteString(fmt.Sprintf("执行时间:%v %s\n", extendData, time.Now().Format("2006-01-02 15:04:05")))
+	if err != nil {
+		fmt.Println("写入文件失败:", err)
+		return err
+	}
+
+	return nil
+}

example/once/main.go

@@ -0,0 +1,257 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/yuninks/timerx"
+)
+
+var save_path = "./once.log"
+
+const (
+	OnceTaskTypeNormal timerx.OnceTaskType = "normal"
+	OnceTaskTypeUrgent timerx.OnceTaskType = "urgent"
+)
+
+func main() {
+
+	ctx := context.Background()
+
+	client := getRedis()
+	once, err := timerx.InitOnce(ctx, client, "once_01", &OnceWorker{}, timerx.WithBatchSize(1000))
+	if err != nil {
+		panic(err)
+	}
+
+	// intervalSaveTime(ctx, once)
+	// intervalSave(ctx, once)
+	// intervalcreateTask(ctx, once)
+	// intervalCreateTime(ctx, once)
+	// benchmarkJob(ctx, once)
+	stabilityTest(ctx, once)
+
+	select {}
+
+}
+
+// 稳定性测试
+func stabilityTest(ctx context.Context, once *timerx.Once) {
+
+	timer := time.NewTicker(time.Second)
+
+	for {
+		select {
+		case t := <-timer.C:
+			fmt.Println("time:", t)
+
+			str := t.Format("2006-01-02 15:04:05")
+
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_0_%d", time.Now().Unix()), 0, "Create 间隔0s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_1_%d", time.Now().Unix()), time.Second*1, "Create 间隔1s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_10_%d", time.Now().Unix()), time.Second*10, "Create 间隔10s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_15_%d", time.Now().Unix()), time.Second*15, "Create 间隔15s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_30_%d", time.Now().Unix()), time.Second*30, "Create 间隔30s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_60_%d", time.Now().Unix()), time.Second*60, "Create 间隔60s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_120_%d", time.Now().Unix()), time.Second*120, "Create 间隔120s ["+str+"]")    // 2分钟
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_300_%d", time.Now().Unix()), time.Second*300, "Create 间隔300s ["+str+"]")    // 5分钟
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_900_%d", time.Now().Unix()), time.Second*900, "Create 间隔900s ["+str+"]")    // 15分钟
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_1800_%d", time.Now().Unix()), time.Second*1800, "Create 间隔1800s ["+str+"]") // 30分钟
+
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_0_%d", time.Now().Unix()), 0, "Save 间隔0s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_1_%d", time.Now().Unix()), time.Second*1, "Save 间隔1s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_10_%d", time.Now().Unix()), time.Second*10, "Save 间隔10s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_15_%d", time.Now().Unix()), time.Second*15, "Save 间隔15s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_30_%d", time.Now().Unix()), time.Second*30, "Save 间隔30s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_60_%d", time.Now().Unix()), time.Second*60, "Save 间隔60s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_120_%d", time.Now().Unix()), time.Second*120, "Save 间隔120s ["+str+"]")    // 2分钟
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_300_%d", time.Now().Unix()), time.Second*300, "Save 间隔300s ["+str+"]")    // 5分钟
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_900_%d", time.Now().Unix()), time.Second*900, "Save 间隔900s ["+str+"]")    // 15分钟
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_1800_%d", time.Now().Unix()), time.Second*1800, "Save 间隔1800s ["+str+"]") // 30分钟
+
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_0_%d", time.Now().Unix()), time.Now(), "CreateByTime 当前时间 ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_1_%d", time.Now().Unix()), time.Now().Add(time.Second*1), "CreateByTime 当前时间+1s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_10_%d", time.Now().Unix()), time.Now().Add(time.Second*10), "CreateByTime 当前时间+10s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_15_%d", time.Now().Unix()), time.Now().Add(time.Second*15), "CreateByTime 当前时间+15s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_30_%d", time.Now().Unix()), time.Now().Add(time.Second*30), "CreateByTime 当前时间+30s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_60_%d", time.Now().Unix()), time.Now().Add(time.Second*60), "CreateByTime 当前时间+60s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_120_%d", time.Now().Unix()), time.Now().Add(time.Second*120), "CreateByTime 当前时间+120s ["+str+"]")    // 2分钟
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_300_%d", time.Now().Unix()), time.Now().Add(time.Second*300), "CreateByTime 当前时间+300s ["+str+"]")    // 5分钟
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_900_%d", time.Now().Unix()), time.Now().Add(time.Second*900), "CreateByTime 当前时间+900s ["+str+"]")    // 15分钟
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_1800_%d", time.Now().Unix()), time.Now().Add(time.Second*1800), "CreateByTime 当前时间+1800s ["+str+"]") // 30分钟
+
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_0_%d", time.Now().Unix()), time.Now(), "SaveByTime 当前时间 ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_1_%d", time.Now().Unix()), time.Now().Add(time.Second*1), "SaveByTime 当前时间+1s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_10_%d", time.Now().Unix()), time.Now().Add(time.Second*10), "SaveByTime 当前时间+10s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_15_%d", time.Now().Unix()), time.Now().Add(time.Second*15), "SaveByTime 当前时间+15s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_30_%d", time.Now().Unix()), time.Now().Add(time.Second*30), "SaveByTime 当前时间+30s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_60_%d", time.Now().Unix()), time.Now().Add(time.Second*60), "SaveByTime 当前时间+60s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_120_%d", time.Now().Unix()), time.Now().Add(time.Second*120), "SaveByTime 当前时间+120s ["+str+"]")    // 2分钟
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_300_%d", time.Now().Unix()), time.Now().Add(time.Second*300), "SaveByTime 当前时间+300s ["+str+"]")    // 5分钟
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_900_%d", time.Now().Unix()), time.Now().Add(time.Second*900), "SaveByTime 当前时间+900s ["+str+"]")    // 15分钟
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_1800_%d", time.Now().Unix()), time.Now().Add(time.Second*1800), "SaveByTime 当前时间+1800s ["+str+"]") // 30分钟
+		}
+	}
+
+}
+
+// 指定时间测试
+func intervalSaveTime(ctx context.Context, once *timerx.Once) {
+
+	beginTime := time.Now()
+
+	for i := 1; i < 100; i++ {
+
+		execTime := beginTime.Add(time.Second * time.Duration(i))
+
+		err := once.SaveByTime(ctx, timerx.OnceTaskType("intervalSaveTime"), fmt.Sprintf("intervalSaveTime_task_%d", i), execTime, fmt.Sprintf("任务数据_%d 预期时间%s", i, execTime.Format("2006-01-02 15:04:05")))
+		fmt.Println(err)
+	}
+
+}
+
+// 间隔测试
+func intervalSave(ctx context.Context, once *timerx.Once) {
+
+	beginTime := time.Now()
+
+	for i := 1; i < 100; i++ {
+
+		execTime := beginTime.Add(time.Second * time.Duration(i))
+
+		err := once.Save(ctx, timerx.OnceTaskType("intervalSaveTime"), fmt.Sprintf("intervalSaveTime_task_%d", i), time.Until(execTime), fmt.Sprintf("任务数据_%d 预期时间%s", i, execTime.Format("2006-01-02 15:04:05")))
+		fmt.Println(err)
+	}
+
+}
+
+// 创建间隔测试
+func intervalcreateTask(ctx context.Context, once *timerx.Once) {
+
+	beginTime := time.Now()
+
+	for i := 1; i < 100; i++ {
+
+		execTime := beginTime.Add(time.Second * time.Duration(i))
+
+		err := once.Create(ctx, timerx.OnceTaskType("intervalSaveTime"), fmt.Sprintf("intervalSaveTime_task_%d", i), time.Until(execTime), fmt.Sprintf("任务数据A_%d 预期时间%s", i, execTime.Format("2006-01-02 15:04:05")))
+		fmt.Println(err)
+		err = once.Create(ctx, timerx.OnceTaskType("intervalSaveTime"), fmt.Sprintf("intervalSaveTime_task_%d", i), time.Until(execTime), fmt.Sprintf("任务数据B_%d 预期时间%s", i, execTime.Format("2006-01-02 15:04:05")))
+		fmt.Println(err)
+	}
+
+}
+
+func intervalCreateTime(ctx context.Context, once *timerx.Once) {
+
+	beginTime := time.Now()
+
+	for i := 1; i < 100; i++ {
+
+		execTime := beginTime.Add(time.Second * time.Duration(i))
+
+		err := once.CreateByTime(ctx, timerx.OnceTaskType("intervalSaveTime"), fmt.Sprintf("intervalSaveTime_task_%d", i), execTime, fmt.Sprintf("任务数据A_%d 预期时间%s", i, execTime.Format("2006-01-02 15:04:05")))
+		fmt.Println(err)
+		err = once.CreateByTime(ctx, timerx.OnceTaskType("intervalSaveTime"), fmt.Sprintf("intervalSaveTime_task_%d", i), execTime, fmt.Sprintf("任务数据B_%d 预期时间%s", i, execTime.Format("2006-01-02 15:04:05")))
+		fmt.Println(err)
+	}
+
+}
+
+// 压力测试
+func benchmarkJob(ctx context.Context, once *timerx.Once) {
+
+	t := time.Now()
+
+	ch := make(chan ChanStatus, 1000)
+
+	go func() {
+		for a := 0; a < 100; a++ {
+			go func(a int) {
+				for status := range ch {
+					// fmt.Println("协程", a, "处理任务", status)
+					// time.Sleep(10 * time.Millisecond) // 模拟处理时间
+					err := once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("task_%d_%d", status.I, status.J), status.T, fmt.Sprintf("任务数据_%d_%d 预期时间%s", status.I, status.J, status.T.Format("2006-01-02 15:04:05")))
+					if err != nil {
+						fmt.Println("保存任务失败:", err)
+					}
+				}
+			}(a)
+		}
+	}()
+
+	go func() {
+		// 一千万任务，每个任务间隔1秒
+
+		for i := 0; i < 100; i++ {
+			runTime := t.Add(time.Duration(i) * time.Second)
+			for j := 0; j < 100; j++ {
+				ch <- ChanStatus{
+					I: i,
+					J: j,
+					T: runTime,
+				}
+			}
+		}
+	}()
+}
+
+type ChanStatus struct {
+	I int
+	J int
+	T time.Time
+}
+
+func getRedis() *redis.Client {
+	client := redis.NewClient(&redis.Options{
+		Addr:     "127.0.0.1" + ":" + "6379",
+		Password: "123456", // no password set
+		DB:       0,        // use default DB
+	})
+	if client == nil {
+		panic("redis init error")
+	}
+	return client
+}
+
+type OnceWorker struct {
+}
+
+func (l *OnceWorker) Worker(ctx context.Context, taskType timerx.OnceTaskType, taskId string, attachData interface{}) *timerx.OnceWorkerResp {
+	// 任务处理逻辑
+	callback(ctx, attachData)
+
+	return &timerx.OnceWorkerResp{}
+}
+
+func callback(ctx context.Context, extendData any) error {
+
+	fmt.Println("任务执行了", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
+
+	// 解析文件路径，每天一个文件
+	path, _ := filepath.Abs("./")
+	// 拼接文件路径
+	save_path = filepath.Join(path, "/cache/once/"+time.Now().Format("2006-01-02")+".log")
+	// 创建文件夹
+	dir := filepath.Dir(save_path)
+	os.MkdirAll(dir, 0755)
+
+	// 追加到文件
+	file, err := os.OpenFile(save_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		fmt.Println("打开文件失败:", err)
+		return err
+	}
+	defer file.Close()
+	_, err = file.WriteString(fmt.Sprintf("执行时间:%v [%s] \n", extendData, time.Now().Format("2006-01-02 15:04:05")))
+	if err != nil {
+		fmt.Println("写入文件失败:", err)
+		return err
+	}
+
+	return nil
+}

example/single/main.go

@@ -0,0 +1,297 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"path/filepath"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/yuninks/timerx"
+)
+
+var save_path string = "./single.log"
+
+func main() {
+
+	ctx := context.Background()
+
+	ops := []timerx.Option{}
+
+	clu := timerx.InitSingle(ctx, ops...)
+
+	space(ctx, clu)
+	minute(ctx, clu)
+	hour(ctx, clu)
+	day(ctx, clu)
+	week(ctx, clu)
+	month(ctx, clu)
+	cron(ctx, clu)
+
+	select {}
+
+}
+
+// space
+func space(ctx context.Context, clu *timerx.Single) {
+	// 每秒执行一次
+	_, err := clu.EverySpace(ctx, "space_test_second_1", 1*time.Second, callback, "space 这是秒任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_second_5", 5*time.Second, callback, "space 这是5秒任务")
+	fmt.Println(err)
+
+	// 每分钟执行一次
+	_, err = clu.EverySpace(ctx, "space_test_minute_1", 1*time.Minute, callback, "space 这是分钟任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_minute_5", 5*time.Minute, callback, "space 这是5分钟任务")
+	fmt.Println(err)
+
+	// 每小时执行一次
+	_, err = clu.EverySpace(ctx, "space_test_hour_1", 1*time.Hour, callback, "space 这是小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_2", 2*time.Hour, callback, "space 这是2小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_3", 3*time.Hour, callback, "space 这是3小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_4", 4*time.Hour, callback, "space 这是4小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_5", 5*time.Hour, callback, "space 这是5小时任务")
+	fmt.Println(err)
+
+	// 每天执行一次
+	_, err = clu.EverySpace(ctx, "space_test_day_1", 24*time.Hour, callback, "space 这是天任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_day_2", 2*24*time.Hour, callback, "space 这是2天任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_day_3", 3*24*time.Hour, callback, "space 这是3天任务")
+	fmt.Println(err)
+
+	// 每周执行一次
+	_, err = clu.EverySpace(ctx, "space_test_week_1", 7*24*time.Hour, callback, "space 这是周任务")
+	fmt.Println(err)
+
+
+	// 每月执行一次
+	_, err = clu.EverySpace(ctx, "space_test_month_1", 30*24*time.Hour, callback, "space 这是月任务")
+	fmt.Println(err)
+
+}
+
+// minute
+func minute(ctx context.Context, clu *timerx.Single) {
+	// 每分钟0s执行一次
+	_, err := clu.EveryMinute(ctx, "minute_test_min1", 0, callback, "minute 这是分钟任务0")
+	fmt.Println(err)
+	// 每分钟5s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min5", 5, callback, "minute 这是分钟任务5")
+	fmt.Println(err)
+	// 每分钟10s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min10", 10, callback, "minute 这是分钟任务10")
+	fmt.Println(err)
+	// 每分钟15s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min15", 15, callback, "minute 这是分钟任务15")
+	fmt.Println(err)
+	// 每分钟30s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min30", 30, callback, "minute 这是分钟任务30")
+	fmt.Println(err)
+	// 每分钟45s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min45", 45, callback, "minute 这是分钟任务45")
+	fmt.Println(err)
+	// 每分钟50s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min50", 50, callback, "minute 这是分钟任务50")
+	fmt.Println(err)
+	// 每分钟55s执行一次
+	_, err = clu.EveryMinute(ctx, "minute_test_min55", 55, callback, "minute 这是分钟任务55")
+	fmt.Println(err)
+}
+
+// Hour
+func hour(ctx context.Context, clu *timerx.Single) {
+	// 每小时的第0分钟15s执行一次
+	_, err := clu.EveryHour(ctx, "hour_test_hour1", 0, 15, callback, "hour 这是小时任务1")
+	fmt.Println(err)
+	// 每小时的第5分钟30s执行一次
+	_, err = clu.EveryHour(ctx, "hour_test_hour2", 5, 30, callback, "hour 这是小时任务2")
+	fmt.Println(err)
+	// 每小时的第10分钟45s执行一次
+	_, err = clu.EveryHour(ctx, "hour_test_hour3", 10, 45, callback, "hour 这是小时任务3")
+	fmt.Println(err)
+	// 每小时的第15分钟0s执行一次
+	_, err = clu.EveryHour(ctx, "hour_test_hour4", 15, 0, callback, "hour 这是小时任务4")
+	fmt.Println(err)
+	// 每小时的第20分钟15s执行一次
+	_, err = clu.EveryHour(ctx, "hour_test_hour5", 20, 15, callback, "hour 这是小时任务5")
+	fmt.Println(err)
+}
+
+// Day
+func day(ctx context.Context, clu *timerx.Single) {
+	// 每天的00:00:00执行一次
+	_, err := clu.EveryDay(ctx, "day_test_day1", 0, 0, 0, callback, "day 这是天任务1")
+	fmt.Println(err)
+	// 每天的02:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day2", 2, 0, 0, callback, "day 这是天任务2")
+	fmt.Println(err)
+	// 每天的04:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day3", 4, 0, 0, callback, "day 这是天任务3")
+	fmt.Println(err)
+	// 每天的06:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day4", 6, 0, 0, callback, "day 这是天任务4")
+	fmt.Println(err)
+	// 每天的08:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day5", 8, 0, 0, callback, "day 这是天任务5")
+	fmt.Println(err)
+	// 每天的10:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day6", 10, 0, 0, callback, "day 这是天任务6")
+	fmt.Println(err)
+	// 每天的12:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day7", 12, 0, 0, callback, "day 这是天任务7")
+	fmt.Println(err)
+	// 每天的14:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day8", 14, 0, 0, callback, "day 这是天任务8")
+	fmt.Println(err)
+	// 每天的16:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day9", 16, 0, 0, callback, "day 这是天任务9")
+	fmt.Println(err)
+	// 每天的18:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day10", 18, 0, 0, callback, "day 这是天任务10")
+	fmt.Println(err)
+	// 每天的20:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day11", 20, 0, 0, callback, "day 这是天任务11")
+	fmt.Println(err)
+	// 每天的22:00:00执行一次
+	_, err = clu.EveryDay(ctx, "day_test_day12", 22, 0, 0, callback, "day 这是天任务12")
+	fmt.Println(err)
+}
+
+// Week
+func week(ctx context.Context, clu *timerx.Single) {
+	// 每周一 10:00:00 执行
+	_, err := clu.EveryWeek(ctx, "week_test_week1", 1, 10, 0, 0, callback, "week 这是周任务1")
+	fmt.Println(err)
+	// 每周二 10:00:00 执行
+	_, err = clu.EveryWeek(ctx, "week_test_week2", 2, 10, 0, 0, callback, "week 这是周任务2")
+	fmt.Println(err)
+	// 每周三 10:00:00 执行
+	_, err = clu.EveryWeek(ctx, "week_test_week3", 3, 10, 0, 0, callback, "week 这是周任务3")
+	fmt.Println(err)
+	// 每周四 10:00:00 执行
+	_, err = clu.EveryWeek(ctx, "week_test_week4", 4, 10, 0, 0, callback, "week 这是周任务4")
+	fmt.Println(err)
+	// 每周五 10:00:00 执行
+	_, err = clu.EveryWeek(ctx, "week_test_week5", 5, 10, 0, 0, callback, "week 这是周任务5")
+	fmt.Println(err)
+	// 每周六 10:00:00 执行
+	_, err = clu.EveryWeek(ctx, "week_test_week6", 6, 10, 0, 0, callback, "week 这是周任务6")
+	fmt.Println(err)
+	// 每周日 10:00:00 执行
+	_, err = clu.EveryWeek(ctx, "week_test_week7", 0, 10, 0, 0, callback, "week 这是周任务7")
+	fmt.Println(err)
+}
+
+// Month
+func month(ctx context.Context, clu *timerx.Single) {
+	// 每月的第1号 10:00:00 执行
+	_, err := clu.EveryMonth(ctx, "month_test_month1", 1, 10, 0, 0, callback, "month 这是月任务1")
+	fmt.Println(err)
+	// 每月的第5号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month5", 5, 10, 0, 0, callback, "month 这是月任务5")
+	fmt.Println(err)
+	// 每月的第10号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month10", 10, 10, 0, 0, callback, "month 这是月任务10")
+	fmt.Println(err)
+	// 每月的第15号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month15", 15, 10, 0, 0, callback, "month 这是月任务15")
+	fmt.Println(err)
+	// 每月的第20号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month20", 20, 10, 0, 0, callback, "month 这是月任务20")
+	fmt.Println(err)
+	// 每月的第25号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month25", 25, 10, 0, 0, callback, "month 这是月任务25")
+	fmt.Println(err)
+	// 每月的第28号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month28", 28, 10, 0, 0, callback, "month 这是月任务28")
+	fmt.Println(err)
+	// 每月的第29号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month29", 29, 10, 0, 0, callback, "month 这是月任务29")
+	fmt.Println(err)
+	// 每月的第30号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month30", 30, 10, 0, 0, callback, "month 这是月任务30")
+	fmt.Println(err)
+	// 每月的第31号 10:00:00 执行
+	_, err = clu.EveryMonth(ctx, "month_test_month31", 31, 10, 0, 0, callback, "month 这是月任务31")
+	fmt.Println(err)
+
+}
+
+func cron(ctx context.Context, clu *timerx.Single) {
+	// 秒级表达式 5秒执行一次
+	_, err := clu.Cron(ctx, "cron_test_cron1", "*/5 * * * * ?", callback, "cron 这是cron任务1", timerx.WithCronParserSecond())
+	fmt.Println(err)
+	// Linux表达式 5分钟执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron2", "*/5 * * * *", callback, "cron 这是cron任务2", timerx.WithCronParserLinux())
+	fmt.Println(err)
+	// 符号表达式 5秒执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron3", "@every 5s", callback, "cron 这是cron任务3", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每天执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron4", "@daily", callback, "cron 这是cron任务4", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每月执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron5", "@monthly", callback, "cron 这是cron任务5", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每年执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron6", "@yearly", callback, "cron 这是cron任务6", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每周执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron7", "@weekly", callback, "cron 这是cron任务7", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每小时执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron8", "@hourly", callback, "cron 这是cron任务8", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+	// 符号表达式 每分钟执行一次
+	_, err = clu.Cron(ctx, "cron_test_cron9", "@minutely", callback, "cron 这是cron任务9", timerx.WithCronParserDescriptor())
+	fmt.Println(err)
+
+}
+
+func getRedis() *redis.Client {
+	client := redis.NewClient(&redis.Options{
+		Addr:     "127.0.0.1" + ":" + "6379",
+		Password: "123456", // no password set
+		DB:       0,        // use default DB
+	})
+	if client == nil {
+		panic("redis init error")
+	}
+	return client
+}
+
+func callback(ctx context.Context, extendData any) error {
+
+	fmt.Println("任务执行了", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
+
+	// 解析文件路径，每天一个文件
+	path, _ := filepath.Abs("./")
+	// 拼接文件路径
+	save_path = filepath.Join(path, "/cache/single/"+time.Now().Format("2006-01-02")+".log")
+	// 创建文件夹
+	dir := filepath.Dir(save_path)
+	os.MkdirAll(dir, 0755)
+
+	// 追加到文件
+	file, err := os.OpenFile(save_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+	if err != nil {
+		fmt.Println("打开文件失败:", err)
+		return err
+	}
+	defer file.Close()
+	_, err = file.WriteString(fmt.Sprintf("执行时间:%v %s\n", extendData, time.Now().Format("2006-01-02 15:04:05")))
+	if err != nil {
+		fmt.Println("写入文件失败:", err)
+		return err
+	}
+
+	return nil
+}

go.mod

@@ -5,9 +5,10 @@ go 1.24
 require (
 	github.com/google/uuid v1.6.0
 	github.com/redis/go-redis/v9 v9.14.0
+	github.com/robfig/cron/v3 v3.0.1
 	github.com/stretchr/testify v1.11.1
-	github.com/yuninks/cachex v1.0.5
-	github.com/yuninks/lockx v1.1.3
+	github.com/yuninks/cachex v1.0.6
+	github.com/yuninks/lockx v1.1.4
 )
 
 require (

go.sum

@@ -19,14 +19,16 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/redis/go-redis/v9 v9.14.0 h1:u4tNCjXOyzfgeLN+vAZaW1xUooqWDqVEsZN0U01jfAE=
 github.com/redis/go-redis/v9 v9.14.0/go.mod h1:huWgSWd8mW6+m0VPhJjSSQ+d6Nh1VICQ6Q5lHuCH/Iw=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
 github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
-github.com/yuninks/cachex v1.0.5 h1:Y2NmTsuEgwEVYb7FVFh5tUN67kmrUioeksQqLbOAwsM=
-github.com/yuninks/cachex v1.0.5/go.mod h1:5357qz18UvHTJSgZzkMamUzZoFzGeKG9+4tIUBXRSVM=
-github.com/yuninks/lockx v1.1.3 h1:OA6rb4/XOj+M+1vKLs8fqsU4ZCadvga+oARTWTHpwx0=
-github.com/yuninks/lockx v1.1.3/go.mod h1:+HyRozwQHMHrykyOFlotV4Z+z2yrgRSdDl8TxxRMFzw=
+github.com/yuninks/cachex v1.0.6 h1:G5dJ8O0gLAGnLwydEHVCSSGv+p1z8KfZdjb5NdxxsVA=
+github.com/yuninks/cachex v1.0.6/go.mod h1:5357qz18UvHTJSgZzkMamUzZoFzGeKG9+4tIUBXRSVM=
+github.com/yuninks/lockx v1.1.4 h1:Wrd4aAU5apNWAamZM7yqoNlwHDjGY/X1YYrLXcgj1+s=
+github.com/yuninks/lockx v1.1.4/go.mod h1:+HyRozwQHMHrykyOFlotV4Z+z2yrgRSdDl8TxxRMFzw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

heartbeat/heartbeat.go

@@ -14,7 +14,7 @@ import (
 )
 
 // 心跳
-// 作用：上报实例存活状态
+// 作用：上报实例最新存活状态,清理过期实例
 // 依赖：leader priority
 
 type HeartBeat struct {

leader/leader.go

@@ -53,6 +53,7 @@ func InitLeader(ctx context.Context, ref redis.UniversalClient, keyPrefix string
 		cancel:           cancel,
 		redis:            ref,
 		leaderUniLockKey: "timer:leader_lockKey" + op.source + keyPrefix,
+		leaderKey:        "timer:leader" + op.source + keyPrefix,
 		priority:         op.priority,
 		instanceId:       op.instanceId,
 		logger:           op.logger,
@@ -61,7 +62,7 @@ func InitLeader(ctx context.Context, ref redis.UniversalClient, keyPrefix string
 	l.wg.Add(1)
 	go l.leaderElection()
 
-	l.logger.Infof(l.ctx, "InitLeader InstanceId %s lockKey:%s", l.instanceId, l.leaderUniLockKey)
+	l.logger.Infof(l.ctx, "InitLeader InstanceId %s lockKey:%s leaderKey:%s", l.instanceId, l.leaderUniLockKey, l.leaderKey)
 
 	return l, nil
 }

next_time.go

@@ -3,6 +3,8 @@ package timerx
 import (
 	"errors"
 	"time"
+
+	"github.com/robfig/cron/v3"
 )
 
 // 计算该任务下次执行时间
@@ -32,6 +34,8 @@ func GetNextTime(t time.Time, job JobData) (*time.Time, error) {
 		next, err = calculateNextMinuteTime(t, job)
 	case JobTypeInterval:
 		next, err = calculateNextInterval(t, job)
+	case JobTypeCron:
+		next, err = calculateNextCronTime(t, job)
 	default:
 		return nil, errors.New("未知的任务类型: " + string(job.JobType))
 	}
@@ -73,6 +77,17 @@ func validateJobData(job JobData) error {
 		if job.BaseTime.IsZero() {
 			return ErrBaseTime
 		}
+	case JobTypeCron:
+		if job.CronExpression == "" {
+			return ErrCronExpression
+		}
+		if job.CronSchedule == nil {
+			return ErrCronParser
+		}
+		_, err := calculateNextCronTime(time.Now(), job)
+		if err != nil {
+			return err
+		}
 	}
 
 	if job.Hour < 0 || job.Hour > 23 {
@@ -216,6 +231,35 @@ func calculateNextMinuteTime(t time.Time, job JobData) (*time.Time, error) {
 	return &nextMinuteTime, nil
 }
 
+// 计算cron任务下下次执行时间
+func calculateNextCronTime(t time.Time, job JobData) (*time.Time, error) {
+	if job.CronExpression == "" {
+		return nil, ErrCronExpression
+	}
+	if job.CronSchedule == nil {
+		return nil, ErrCronParser
+	}
+
+	s := *job.CronSchedule
+
+	next := s.Next(t)
+	return &next, nil
+}
+
+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
+}
+
 // 检查是否本周期可以运行
 // 检查是否本周期可以运行（已弃用，使用新的时间比较逻辑）
 // 保留此函数用于向后兼容，但建议使用新的时间计算逻辑

next_time_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/robfig/cron/v3"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -12,6 +13,15 @@ func TestGetNextTime(t *testing.T) {
 
 	tt := time.Date(2025, 10, 16, 10, 30, 5, 0, time.Local)
 
+	cronExpression := "0 0 10 * * ?" // Every day at 10:00 AM
+	parser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
+	sche, err := GetCronSche(cronExpression, &parser)
+	if err != nil {
+		t.Fatal(err)
+	}
+	// ttt := (*sche).Next(tt)
+	// fmt.Println(ttt.Format("2006-01-02 15:04:05"))
+
 	// Test cases
 	tests := []struct {
 		name          string
@@ -103,6 +113,15 @@ func TestGetNextTime(t *testing.T) {
 			expectedTime:  tt.Add(1 * time.Second), // Assuming current date is March 7, 2022, 10:30 AM
 			expectedError: nil,
 		},
+		{
+			name: "Test JobTypeCron",
+			job: JobData{
+				JobType:        JobTypeCron,
+				CronExpression: cronExpression,
+				CronSchedule:   sche,
+			},
+			expectedTime: time.Date(2025, 10, 17, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
+		},
 		{
 			name: "Test unknown JobType",
 			job: JobData{

once.go

@@ -6,13 +6,14 @@ import (
 	"errors"
 	"fmt"
 	"runtime/debug"
+	"sort"
 	"strconv"
 	"strings"
 	"sync"
 	"time"
 
-	"github.com/redis/go-redis/v9"
 	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
 	"github.com/yuninks/lockx"
 	"github.com/yuninks/timerx/heartbeat"
 	"github.com/yuninks/timerx/leader"
@@ -51,7 +52,9 @@ type Once struct {
 	keySeparator   string        // 分割符
 	timeout        time.Duration // 任务执行超时时间
 
-	maxRetryCount int // 最大重试次数 0代表不限
+	maxRunCount int           // 最大重试次数 0代表不限
+	workerChan  chan struct{} // worker
+	maxWorkers  int           // 最大worker数量
 }
 
 type OnceWorkerResp struct {
@@ -74,11 +77,19 @@ type Callback interface {
 }
 
 type extendData struct {
-	Delay      time.Duration
+	TaskTimes []time.Time
 	Data      any
-	RetryCount int // 重试次数
+	RunCount  int // 运行次数
+	JobType   jobType
 }
 
+type jobType string
+
+const (
+	jobTypeOnce = "once"
+	jobTypeList = "list"
+)
+
 // 初始化
 func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, call Callback, opts ...Option) (*Once, error) {
 	op := newOptions(opts...)
@@ -100,7 +111,7 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 		listKey:          "timer:once_listkey" + keyPrefix,
 		executeInfoKey:   "timer:once_executeInfoKey" + keyPrefix,
 		globalLockPrefix: "timer:once_globalLockPrefix" + keyPrefix,
-		usePriority:      op.usePriority,
+		usePriority:      false,
 		redis:            re,
 		worker:           call,
 		keyPrefix:        keyPrefix,
@@ -109,17 +120,33 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 		instanceId:       u.String(),
 		keySeparator:     "[:]",
 		timeout:          op.timeout,
-		maxRetryCount:    op.maxRetryCount,
+		maxRunCount:      op.maxRunCount,
+		workerChan:       make(chan struct{}, op.maxWorkers),
+		maxWorkers:       op.maxWorkers,
 	}
 
 	// 初始化优先级
-	if wo.usePriority {
+	if op.priorityType != priorityTypeNone {
+
+		wo.usePriority = true
+
+		if op.priorityType == priorityTypeVersion {
+			pVal, err := priority.PriorityByVersion(op.priorityVersion)
+			if err != nil {
+				wo.logger.Errorf(ctx, "PriorityByVersion version:%s err:%v", op.priorityVersion, err)
+				return nil, err
+			}
+			op.priorityVal = pVal
+		}
+
 		pri, err := priority.InitPriority(
 			ctx,
 			re,
 			keyPrefix,
 			op.priorityVal,
 			priority.WithLogger(wo.logger),
+			priority.WithInstanceId(wo.instanceId),
+			priority.WithSource("once"),
 		)
 		if err != nil {
 			wo.logger.Errorf(ctx, "InitPriority err:%v", err)
@@ -136,6 +163,7 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 		leader.WithLogger(wo.logger),
 		leader.WithPriority(wo.priority),
 		leader.WithInstanceId(wo.instanceId),
+		leader.WithSource("once"),
 	)
 	if err != nil {
 		wo.logger.Infof(ctx, "InitLeader err:%v", err)
@@ -261,19 +289,21 @@ func (l *Once) executeTasks() {
 
 	for {
 
-		if l.usePriority {
-			if !l.priority.IsLatest(l.ctx) {
-				time.Sleep(time.Second * 5)
-				continue
-			}
-		}
-
 		select {
 		case <-l.stopChan:
 			return
 		case <-l.ctx.Done():
 			return
-		default:
+		case l.workerChan <- struct{}{}:
+			func() {
+				defer func() {
+					<-l.workerChan
+				}()
+
+				if l.usePriority && !l.priority.IsLatest(l.ctx) {
+					time.Sleep(time.Second * 5)
+					return
+				}
 
 				keys, err := l.redis.BLPop(l.ctx, time.Second*10, l.listKey).Result()
 				if err != nil {
@@ -282,11 +312,17 @@ func (l *Once) executeTasks() {
 						// Redis 异常，休眠一会儿再重试
 						time.Sleep(time.Second * 5)
 					}
-				continue
+					return
 				}
 
+				if len(keys) < 2 {
+					l.logger.Errorf(l.ctx, "Invalid task data: %v", keys)
+					// 数据异常，继续下一个
+					return
+				}
+				// 处理任务
 				go l.processTask(keys[1])
-
+			}()
 		}
 	}
 
@@ -312,40 +348,77 @@ func (l *Once) parseRedisKey(key string) (OnceTaskType, string, error) {
 // @param uniTaskId string 任务唯一标识
 // @param delayTime time.Duration 延迟时间
 // @param attachData interface{} 附加数据
-func (l *Once) Save(taskType OnceTaskType, taskId string, delayTime time.Duration, attachData interface{}) error {
-	return l.save(taskType, taskId, delayTime, attachData, 0)
+func (l *Once) Save(ctx context.Context, taskType OnceTaskType, taskId string, delayTime time.Duration, attachData any) error {
+	if delayTime < 0 {
+		delayTime = 0
+	}
+	execTime := time.Now().Add(delayTime)
+	return l.save(ctx, jobTypeOnce, taskType, taskId, []time.Time{execTime}, attachData, 0)
+}
+
+// 指定时间添加任务(覆盖)
+func (l *Once) SaveByTime(ctx context.Context, taskType OnceTaskType, taskId string, executeTime time.Time, attachData any) error {
+	return l.save(ctx, jobTypeOnce, taskType, taskId, []time.Time{executeTime}, attachData, 0)
+}
+
+func (l *Once) SaveByList(ctx context.Context, taskType OnceTaskType, taskId string, executeTimes []time.Time, attachData any, runCount int) error {
+	return l.save(ctx, jobTypeList, taskType, taskId, executeTimes, attachData, runCount)
 }
 
 // 添加任务(覆盖)
 // 重复插入就代表覆盖
-func (w *Once) save(taskType OnceTaskType, taskId string, delayTime time.Duration, attachData interface{}, retryCount int) error {
-	if delayTime <= 0 {
-		return fmt.Errorf("delay time must be positive")
+func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType, taskId string, taskTimes []time.Time, attachData any, runCount int) error {
+	if len(taskTimes) == 0 {
+		w.logger.Errorf(ctx, "delay time must be positive taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount)
+		return ErrExecuteTime
+	}
+
+	if jobType == jobTypeList && len(taskTimes) <= runCount {
+		w.logger.Errorf(ctx, "delay time must be positive taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount)
+		return ErrRunCount
+	}
+
+	// 根据时间从小到大排序
+	sort.Slice(taskTimes, func(i, j int) bool {
+		return taskTimes[i].Before(taskTimes[j])
+	})
+
+	latestTime := taskTimes[len(taskTimes)-1]
+	if latestTime.Before(time.Now()) {
+		w.logger.Errorf(ctx, "delay time must be positive taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount)
+		return ErrDelayTime
+	}
+
+	nextTime := taskTimes[0]
+	if jobType == jobTypeList {
+		nextTime = taskTimes[runCount]
 	}
 
 	redisKey := w.buildRedisKey(taskType, taskId)
-	executeTime := time.Now().Add(delayTime)
 
 	ed := extendData{
-		Delay:      delayTime,
+		TaskTimes: taskTimes,
 		Data:      attachData,
-		RetryCount: retryCount,
+		RunCount:  runCount,
+		JobType:   jobType,
 	}
 	b, _ := json.Marshal(ed)
 
 	// 使用事务确保原子性
 	pipe := w.redis.TxPipeline()
 
-	dataExpire := delayTime + time.Minute*30
+	expiresTime := latestTime.Add(time.Minute * 30)
+
+	dataExpire := time.Until(expiresTime)
 
 	pipe.SetEx(w.ctx, w.keyPrefix+redisKey, b, dataExpire)
 	pipe.ZAdd(w.ctx, w.zsetKey, redis.Z{
-		Score:  float64(executeTime.UnixMilli()),
+		Score:  float64(nextTime.UnixMilli()),
 		Member: redisKey,
 	})
 	_, err := pipe.Exec(w.ctx)
 	if err != nil {
-		w.logger.Errorf(w.ctx, "save task failed:%v", err)
+		w.logger.Errorf(w.ctx, "save task failed:%v taskType:%v taskId:%v attachData:%v retryCount:%v", err, taskType, taskId, attachData, runCount)
 		return err
 	}
 
@@ -353,13 +426,28 @@ func (w *Once) save(taskType OnceTaskType, taskId string, delayTime time.Duratio
 }
 
 // 添加任务(不覆盖)
-func (l *Once) Create(taskType OnceTaskType, taskId string, delayTime time.Duration, attachData any) error {
-	return l.create(taskType, taskId, delayTime, attachData, 0)
+func (l *Once) Create(ctx context.Context, taskType OnceTaskType, taskId string, delayTime time.Duration, attachData any) error {
+	if delayTime <= 0 {
+		delayTime = 0
+	}
+	execTime := time.Now().Add(delayTime)
+	return l.create(ctx, jobTypeOnce, taskType, taskId, []time.Time{execTime}, attachData, 0)
 }
 
-func (l *Once) create(taskType OnceTaskType, taskId string, delayTime time.Duration, attachData any, retryCount int) error {
-	if delayTime <= 0 {
-		return fmt.Errorf("delay time must be positive")
+// 指定时间执行(不覆盖)
+func (l *Once) CreateByTime(ctx context.Context, taskType OnceTaskType, taskId string, executeTime time.Time, attachData any) error {
+	return l.create(ctx, jobTypeOnce, taskType, taskId, []time.Time{executeTime}, attachData, 0)
+}
+
+// 指定多个时间执行(不覆盖)
+func (l *Once) CreateByList(ctx context.Context, taskType OnceTaskType, taskId string, taskTimes []time.Time, attachData any) error {
+	return l.create(ctx, jobTypeList, taskType, taskId, taskTimes, attachData, 0)
+}
+
+func (l *Once) create(ctx context.Context, jobType jobType, taskType OnceTaskType, taskId string, taskTimes []time.Time, attachData any, runCount int) error {
+	if len(taskTimes) <= 0 {
+		l.logger.Errorf(ctx, "delay time must be positive taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount)
+		return ErrExecuteTime
 	}
 
 	redisKey := l.buildRedisKey(taskType, taskId)
@@ -367,16 +455,17 @@ func (l *Once) create(taskType OnceTaskType, taskId string, delayTime time.Durat
 	score, err := l.redis.ZScore(l.ctx, l.zsetKey, redisKey).Result()
 	if err != nil {
 		if errors.Is(err, redis.Nil) {
-			return l.Save(taskType, taskId, delayTime, attachData)
+			return l.save(ctx, jobType, taskType, taskId, taskTimes, attachData, runCount)
 		}
 		l.logger.Errorf(l.ctx, "redis.ZScore err:%v", err)
 		return err
 	}
 	if score > 0 {
-		return fmt.Errorf("task already exists")
+		l.logger.Errorf(l.ctx, "task exists taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount)
+		return ErrTaskExists
 	}
 
-	return l.save(taskType, taskId, delayTime, attachData, retryCount)
+	return l.save(ctx, jobType, taskType, taskId, taskTimes, attachData, runCount)
 }
 
 // 删除任务
@@ -433,6 +522,7 @@ func (l *Once) batchGetTasks() {
 
 // 执行任务
 func (l *Once) processTask(key string) {
+
 	begin := time.Now()
 
 	ctx, cancel := context.WithTimeout(l.ctx, l.timeout)
@@ -508,23 +598,24 @@ func (l *Once) processTask(key string) {
 func (l *Once) handleRetry(ctx context.Context, taskType OnceTaskType, taskId string,
 	ed *extendData, resp *OnceWorkerResp) error {
 	// 限制重试次数
-	ed.RetryCount++
-	if l.maxRetryCount > 0 && ed.RetryCount > l.maxRetryCount {
-		l.logger.Infof(ctx, "handleRetry task exceeded retry limit: %s %s %d", taskType, taskId, l.maxRetryCount)
+	ed.RunCount++
+	if l.maxRunCount > 0 && ed.RunCount > l.maxRunCount {
+		l.logger.Infof(ctx, "handleRetry task exceeded retry limit: %s %s %d", taskType, taskId, l.maxRunCount)
 		return nil
 	}
 
 	// 更新延迟时间
-	if resp.DelayTime > 0 {
-		ed.Delay = resp.DelayTime
+	if ed.JobType == jobTypeOnce {
+		ed.TaskTimes = []time.Time{time.Now().Add(resp.DelayTime)}
 	}
+
 	if resp.AttachData != nil {
 		ed.Data = resp.AttachData
 	}
 
 	l.logger.Infof(ctx, "handleRetry retrying task: %s:%s, retry count: %d",
-		taskType, taskId, ed.RetryCount)
+		taskType, taskId, ed.RunCount)
 
 	// 不覆盖的新建
-	return l.create(taskType, taskId, ed.Delay, ed.Data, ed.RetryCount)
+	return l.create(ctx, ed.JobType, taskType, taskId, ed.TaskTimes, ed.Data, ed.RunCount)
 }

option.go

@@ -3,6 +3,7 @@ package timerx
 import (
 	"time"
 
+	"github.com/robfig/cron/v3"
 	"github.com/yuninks/timerx/logger"
 )
 
@@ -10,26 +11,44 @@ type Options struct {
 	logger          logger.Logger
 	location        *time.Location
 	timeout         time.Duration // 任务最长执行时间
-	usePriority   bool
-	priorityVal   int64
+	priorityType    priorityType  // 策略类型 0.不使用 1.优先级 2.版本
+	priorityVal     int64         // 策略优先级
+	priorityVersion string        // 策略版本的集
 	batchSize       int
-	maxRetryCount int
+	maxRunCount     int          // 单个任务最大运行次数 0代表不限
+	maxWorkers      int          // 最大工作协程数
+	cronParser      *cron.Parser // cron表达式解析器
 }
 
+type priorityType int8
+
+const (
+	priorityTypeNone     priorityType = 0 // 不使用优先级
+	priorityTypePriority priorityType = 1
+	priorityTypeVersion  priorityType = 2 // 版本
+)
+
 func defaultOptions() Options {
+
+	// 默认使用Linux的定时任务兼容
+	parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
+
 	return Options{
 		logger:       logger.NewLogger(),
 		location:     time.Local,
 		timeout:      time.Hour, //
-		usePriority:   false,
+		priorityType: priorityTypeNone,
 		priorityVal:  0,
 		batchSize:    100,
-		maxRetryCount: 0,
+		maxRunCount:  0,
+		maxWorkers:   100,
+		cronParser:   &parser,
 	}
 }
 
 type Option func(*Options)
 
+// 返回带默认值的配置
 func newOptions(opts ...Option) Options {
 	o := defaultOptions()
 	for _, opt := range opts {
@@ -38,6 +57,15 @@ func newOptions(opts ...Option) Options {
 	return o
 }
 
+// 返回空的配置
+func newEmptyOptions(opts ...Option) Options {
+	o := Options{}
+	for _, opt := range opts {
+		opt(&o)
+	}
+	return o
+}
+
 // 设置日志
 func WithLogger(log logger.Logger) Option {
 	return func(o *Options) {
@@ -62,11 +90,18 @@ func WithTimeout(d time.Duration) Option {
 // 设置优先级
 func WithPriority(priority int64) Option {
 	return func(o *Options) {
-		o.usePriority = true
+		o.priorityType = priorityTypePriority
 		o.priorityVal = priority
 	}
 }
 
+func WithPriorityByVersion(version string) Option {
+	return func(o *Options) {
+		o.priorityType = priorityTypeVersion
+		o.priorityVersion = version
+	}
+}
+
 func WithBatchSize(size int) Option {
 	return func(o *Options) {
 		if size <= 1 {
@@ -81,6 +116,68 @@ func WithMaxRetryCount(count int) Option {
 		if count < 0 {
 			count = 0
 		}
-		o.maxRetryCount = count
+		o.maxRunCount = count
+	}
+}
+
+func WithMaxWorkers(count int) Option {
+	return func(o *Options) {
+		if count < 0 {
+			count = 10
+		}
+		o.maxWorkers = count
+	}
+}
+
+// 添加cron表达式解析器
+func WithCronParser(parser cron.Parser) Option {
+	return func(o *Options) {
+		o.cronParser = &parser
+	}
+}
+
+// 设置cron表达式解析器 秒级
+// "*/5 * * * * ?" => 每隔5秒执行一次
+// "0 0 0 * * ?" => 每天零点执行一次
+// "0 0 0 1 * ?" => 每月1日零点执行一次
+// "0 */5 * * * ?" => 每隔5分钟执行一次
+func WithCronParserSecond() Option {
+	return func(o *Options) {
+		parser := cron.NewParser(cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor)
+		o.cronParser = &parser
+	}
+}
+
+// 设置cron表达式解析器
+// cron.Second | cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow | cron.Descriptor
+func WithCronParserOption(options cron.ParseOption) Option {
+	return func(o *Options) {
+		parser := cron.NewParser(options)
+		o.cronParser = &parser
+	}
+}
+
+// Cron表达式 与Linux的定时任务兼容
+func WithCronParserLinux() Option {
+	return func(o *Options) {
+		parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
+		o.cronParser = &parser
+	}
+}
+
+// Cron表达式 符号
+// @yearly @annually => 每年执行一次，等同于 "0 0 0 1 1 *"
+// @monthly => 每月执行一次，等同于 "0 0 0 1 * *"
+// @weekly => 每周执行一次，等同于 "0 0 0 * * 0"
+// @daily @midnight => 每天执行一次，等同于 "0 0 0 * * *"
+// @hourly => 每小时执行一次，等同于 "0 0 * * * *"
+// @minutely => 每分钟执行一次，等同于 "0 * * * * *"
+// @secondly => 每秒执行一次，等同于 "* * * * * *"
+// @every(time.Duration) => 每隔指定时间执行一次，等同于 "@every 5s"
+
+func WithCronParserDescriptor() Option {
+	return func(o *Options) {
+		parser := cron.NewParser(cron.Descriptor)
+		o.cronParser = &parser
 	}
 }

priority/version_test.go

@@ -6,7 +6,6 @@ import (
 	"github.com/yuninks/timerx/priority"
 )
 
-
 func TestVersionToPriority(t *testing.T) {
 	tests := []struct {
 		name    string
@@ -20,6 +19,18 @@ func TestVersionToPriority(t *testing.T) {
 			want:    1002003000000,
 			wantErr: false,
 		},
+		{
+			name:    "standard version0",
+			version: "0.0.0",
+			want:    0,
+			wantErr: false,
+		},
+				{
+			name:    "standard version1",
+			version: "1.0.0",
+			want:    1000000000000,
+			wantErr: false,
+		},
 		{
 			name:    "version with v prefix",
 			version: "v1.2.3",

readme.md

@@ -1,36 +1,29 @@
 # 功能支持
 
-1. 支持本地任务
-2. 支持集群任务
-3. 支持单次任务
+1. [X] 支持本地任务
+2. [X] 支持集群任务
+3. [X] 支持单次任务
 
 # 功能说明
 
-
-
 # 功能实现
 
 1. 集群间任务调度和任务的唯一依赖于redis进行实现
 
-
 # 缺陷
+
 1. 针对月的任务，需要注意日期有效性，且在月末的最后一天，需要考虑月末的最后一天的下一个任务执行时间
+2. 集群部署时，存在新旧的代码混合问题，任务调度可能存在问题（需要根据实际需要进行版本上线/下线操作）
+3. 主从切换时也要做到平滑上下线
 
-
-1. 集群部署时，存在新旧的代码混合问题，任务调度可能存在问题（需要根据实际需要进行版本上线/下线操作）
 ## 方案一
+
 1. 启动的时候定时向redis注册任务项
 2. 每次计算执行时间的时候根据注册的任务项进行任务计算
 3. 注册任务项需要有下线机制，避免能运行它的节点下线了它还被执行
 
-
 现在有根据要求根据系统时间整点运行任务的要求，这个比简单的定时重复更复杂，因为不但要按时执行，并且不能重复执行，需要全局记录任务执行的状态，由于任务的间隔时间不确定，这个任务执行状态的保存周期也是有变化的
 
 # 待实现
+
 - [ ] 允许执行完重置任务倒计时
-
-
-
-
-
-

single.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/robfig/cron/v3"
 	"github.com/yuninks/timerx/logger"
 )
 
@@ -26,11 +27,12 @@ type Single struct {
 	nextTime    time.Time
 	nextTimeMux sync.RWMutex
 	wg          sync.WaitGroup
-	workerList  sync.Map
-	timerIndex  int64
-	stopChan    chan struct{}
-	hasRun      sync.Map
-	timeout     time.Duration
+	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表达式解析器
 }
 
 // 定时器类
@@ -48,6 +50,7 @@ func InitSingle(ctx context.Context, opts ...Option) *Single {
 		nextTime:   time.Now(),
 		stopChan:   make(chan struct{}),
 		timeout:    op.timeout,
+		cronParser: op.cronParser,
 	}
 
 	sin.startDaemon()
@@ -249,15 +252,14 @@ func (c *Single) EveryMinute(ctx context.Context, taskId string, second int, cal
 
 // 特定时间间隔
 func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
-	nowTime := time.Now().In(c.location)
 
 	if spaceTime < 0 {
 		c.logger.Errorf(ctx, "间隔时间不能小于0")
 		return 0, ErrIntervalTime
 	}
 
-	// 获取当天的零点时间
-	zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, c.location)
 
 	jobData := JobData{
 		JobType: JobTypeInterval,
@@ -270,6 +272,35 @@ func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.D
 	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 回调函数

types.go

@@ -3,6 +3,8 @@ package timerx
 import (
 	"context"
 	"time"
+
+	"github.com/robfig/cron/v3"
 )
 
 type timerStr struct {
@@ -23,6 +25,7 @@ const (
 	JobTypeEveryMinute JobType = "every_minute" // 每分钟
 	JobTypeEverySecond JobType = "every_second" // 每秒
 	JobTypeInterval    JobType = "interval"     // 指定时间间隔
+	JobTypeCron        JobType = "cron"         // cron表达式
 )
 
 type JobData struct {
@@ -37,6 +40,8 @@ type JobData struct {
 	Hour           int            // 每天的第几个小时
 	Minute         int            // 每小时的第几分钟
 	Second         int            // 每分钟的第几秒
+	CronExpression string         // cron表达式
+	CronSchedule   *cron.Schedule // cron表达式解析后的数据
 }
 
 // 定义各个回调函数