修改附加内容的字段为字节

cluster.go

@@ -5,16 +5,18 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"sync"
 	"time"
 
-	"code.yun.ink/pkg/lockx"
 	"github.com/go-redis/redis/v8"
+	uuid "github.com/satori/go.uuid"
+	"github.com/yuninks/cachex"
+	"github.com/yuninks/lockx"
 )
 
 // 功能描述
+
 // 这是基于Redis的定时任务调度器，能够有效的在服务集群里面调度任务，避免了单点压力过高或单点故障问题
 // 由于所有的服务代码是一致的，也就是一个定时任务将在所有的服务都有注册，具体调度到哪个服务运行看调度结果
 
@@ -27,33 +29,52 @@ var clusterOnceLimit sync.Once
 var clusterWorkerList sync.Map
 
 type Cluster struct {
-	ctx     context.Context
-	redis   *redis.Client
+	ctx       context.Context
+	redis     redis.UniversalClient
+	cache     *cachex.Cache
+	timeout   time.Duration
+	logger    Logger
+	keyPrefix string         // key前缀
+	location  *time.Location // 根据时区计算的时间
+
 	lockKey string // 全局计算的key
-	nextKey string // 下一次执行的key
 	zsetKey string // 有序集合的key
 	listKey string // 可执行的任务列表的key
+	setKey  string // 重入集合的key
 }
 
 var clu *Cluster = nil
 
-func InitCluster(ctx context.Context, red *redis.Client) *Cluster {
+// 初始化定时器
+// 全局只需要初始化一次
+func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix string, opts ...Option) *Cluster {
+
 	clusterOnceLimit.Do(func() {
+		op := newOptions(opts...)
+
 		clu = &Cluster{
-			ctx:     ctx,
-			redis:   red,
-			lockKey: "timer:cluster_globalLockKey", // 定时器的全局锁
-			nextKey: "timer:cluster_nextKey",
-			zsetKey: "timer:cluster_zsetKey",
-			listKey: "timer:cluster_listKey",
+			ctx:       ctx,
+			redis:     red,
+			cache:     cachex.NewCache(),
+			timeout:   op.timeout,
+			logger:    op.logger,
+			keyPrefix: keyPrefix,
+			location:  op.location,
+			lockKey:   "timer:cluster_globalLockKey" + keyPrefix, // 定时器的全局锁
+			zsetKey:   "timer:cluster_zsetKey" + keyPrefix,       // 有序集合
+			listKey:   "timer:cluster_listKey" + keyPrefix,       // 列表
+			setKey:    "timer:cluster_setKey" + keyPrefix,        // 重入集合
 		}
 
+		// 设置锁的超时时间
+		lockx.InitOption(lockx.SetTimeout(op.timeout))
+
 		// 监听任务
 		go clu.watch()
 
 		timer := time.NewTicker(time.Millisecond * 200)
 
-		go func(ctx context.Context, red *redis.Client) {
+		go func(ctx context.Context) {
 		Loop:
 			for {
 				select {
@@ -64,74 +85,171 @@ func InitCluster(ctx context.Context, red *redis.Client) *Cluster {
 					break Loop
 				}
 			}
-		}(ctx, red)
+		}(ctx)
 	})
 	return clu
 }
 
-func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Duration, callback callback, extendData interface{}) error {
-	_, ok := clusterWorkerList.Load(uniqueKey)
+// 每月执行一次
+// @param ctx 上下文
+// @param taskId 任务ID
+// @param day 每月的几号
+// @param hour 小时
+// @param minute 分钟
+// @param second 秒
+// @param callback 回调函数
+// @param extendData 扩展数据
+// @return error
+func (c *Cluster) EveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+	nowTime := time.Now().In(c.location)
+
+	jobData := JobData{
+		JobType:    JobTypeEveryMonth,
+		CreateTime: nowTime,
+		Day:        day,
+		Hour:       hour,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, taskId, jobData, callback, extendData)
+}
+
+// 每周执行一次
+// @param ctx context.Context 上下文
+// @param taskId string 任务ID
+// @param week time.Weekday 周
+// @param hour int 小时
+// @param minute int 分钟
+// @param second int 秒
+func (c *Cluster) EveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+	nowTime := time.Now().In(c.location)
+
+	jobData := JobData{
+		JobType:    JobTypeEveryWeek,
+		CreateTime: nowTime,
+		Weekday:    week,
+		Hour:       hour,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, taskId, jobData, callback, extendData)
+}
+
+// 每天执行一次
+func (c *Cluster) EveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+	nowTime := time.Now().In(c.location)
+
+	jobData := JobData{
+		JobType:    JobTypeEveryDay,
+		CreateTime: nowTime,
+		Hour:       hour,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, taskId, jobData, callback, extendData)
+}
+
+// 每小时执行一次
+func (c *Cluster) EveryHour(ctx context.Context, taskId string, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+	nowTime := time.Now().In(c.location)
+
+	jobData := JobData{
+		JobType:    JobTypeEveryHour,
+		CreateTime: nowTime,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, taskId, jobData, callback, extendData)
+}
+
+// 每分钟执行一次
+func (c *Cluster) EveryMinute(ctx context.Context, taskId string, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+	nowTime := time.Now().In(c.location)
+
+	jobData := JobData{
+		JobType:    JobTypeEveryMinute,
+		CreateTime: nowTime,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, taskId, 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())
+
+	jobData := JobData{
+		JobType:      JobTypeInterval,
+		BaseTime:     zeroTime, // 默认当天的零点
+		CreateTime:   nowTime,
+		IntervalTime: spaceTime,
+	}
+
+	return c.addJob(ctx, taskId, jobData, callback, extendData)
+}
+
+// 统一添加任务
+// @param ctx context.Context 上下文
+// @param taskId string 任务ID
+// @param jobData *JobData 任务数据
+// @param callback callback 回调函数
+// @param extendData interface{} 扩展数据
+// @return error
+func (c *Cluster) addJob(ctx context.Context, taskId string, jobData JobData, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
+	_, ok := clusterWorkerList.Load(taskId)
 	if ok {
+		c.logger.Errorf(ctx, "key已存在:%s", taskId)
 		return errors.New("key已存在")
 	}
 
-	if spaceTime != spaceTime.Abs() {
-		return errors.New("时间间隔不能为负数")
+	_, err := GetNextTime(time.Now().In(c.location), jobData)
+	if err != nil {
+		c.logger.Errorf(ctx, "获取下次执行时间失败:%s", err.Error())
+		return err
 	}
 
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
+	// ctx, cancel := context.WithCancel(ctx)
+	// defer cancel()
 
-	lock := lockx.NewGlobalLock(ctx, c.redis, uniqueKey)
-	tB := lock.Try(10)
-	if !tB {
-		return errors.New("添加失败")
-	}
-	defer lock.Unlock()
-
-	nowTime := time.Now()
+	// lock := lockx.NewGlobalLock(ctx, c.redis, taskId)
+	// tB := lock.Try(2)
+	// if !tB {
+	// 	c.logger.Errorf(ctx, "添加失败:%s", taskId)
+	// 	return errors.New("添加失败")
+	// }
+	// defer lock.Unlock()
 
 	t := timerStr{
-		BeginTime:  nowTime,
-		NextTime:   nowTime,
-		SpaceTime:  spaceTime,
 		Callback:   callback,
 		ExtendData: extendData,
-		UniqueKey:  uniqueKey,
+		TaskId:     taskId,
+		JobData:    &jobData,
 	}
 
-	clusterWorkerList.Store(uniqueKey, t)
-
-	cacheStr, _ := c.redis.Get(ctx, c.nextKey).Result()
-	execTime := make(map[string]time.Time)
-	json.Unmarshal([]byte(cacheStr), &execTime)
-
-	p := c.redis.Pipeline()
-
-	p.ZAdd(ctx, c.zsetKey, &redis.Z{
-		Score:  float64(nextTime.UnixMilli()),
-		Member: uniqueKey,
-	})
-	execTime[uniqueKey] = nextTime
-	n, _ := json.Marshal(execTime)
-	// fmt.Println("execTime:", execTime, string(n))
-	p.Set(ctx, c.nextKey, string(n), 0)
-
-	_, err := p.Exec(ctx)
-
-	// fmt.Println("添加", err)
+	clusterWorkerList.Store(taskId, t)
 
 	return err
 }
 
 // 计算下一次执行的时间
+// TODO:注册的任务需放在Redis集中存储，因为本地的话，如果有多个服务，那么就会出现不一致的情况。但是要注意服务如何进行下线，由于是主动上报的，需要有一个机制进行删除过期的任务（添加任务&定时器轮训注册）
+// TODO:考虑不同实例系统时间不一样，可能计算的下次时间不一致，会有重复执行的可能
 func (c *Cluster) getNextTime() {
 
-	// log.Println("begin computer")
-	ctx, cancel := context.WithCancel(c.ctx)
-	defer cancel()
-
-	lock := lockx.NewGlobalLock(ctx, c.redis, c.lockKey)
+	lock := lockx.NewGlobalLock(c.ctx, c.redis, c.lockKey)
 	// 获取锁
 	lockBool := lock.Lock()
 	if !lockBool {
@@ -142,51 +260,73 @@ func (c *Cluster) getNextTime() {
 
 	// 计算下一次时间
 
-	// 读取执行的缓存
-	cacheStr, _ := c.redis.Get(ctx, c.nextKey).Result()
-	execTime := make(map[string]time.Time)
-	json.Unmarshal([]byte(cacheStr), &execTime)
-
-	p := c.redis.Pipeline()
-
-	nowTime := time.Now()
+	// p := c.redis.Pipeline()
 
+	// 根据内部注册的任务列表计算下一次执行的时间
 	clusterWorkerList.Range(func(key, value interface{}) bool {
 		val := value.(timerStr)
-		beforeTime := execTime[val.UniqueKey]
-		if beforeTime.After(nowTime) {
+
+		nextTime, _ := GetNextTime(time.Now().In(c.location), *val.JobData)
+
+		// fmt.Println(val.ExtendData, val.JobData, nextTime)
+
+		// 内部判定是否重复
+		cacheKey := fmt.Sprintf("%s_%s_%d", c.keyPrefix, val.TaskId, nextTime.UnixMilli())
+		cacheVal, err := c.cache.Get(cacheKey)
+		if err == nil {
+			// 缓存已有值
 			return true
 		}
-		nextTime := getNextExecTime(beforeTime, val.SpaceTime)
-		execTime[val.UniqueKey] = nextTime
+		valueNum := int(0)
+		if cacheVal != nil {
+			valueNum = cacheVal.(int)
+		}
+		if valueNum > 2 {
+			// 重试2次还是失败就不执行了
+			return true
+		}
+		// fmt.Println("计算时间1", val.ExtendData, time.UnixMilli(nextTime.UnixMilli()).Format("2006-01-02 15:04:05"))
+
+		// redis lua脚本，尝试设置nx锁时间为一分钟，如果能设置进去则添加到有序集合zsetKey
+		script := `
+		local zsetKey = KEYS[1]
+
+		local cacheKey = ARGV[1]
+		local expireTime = ARGV[2]
+	
+		local score = ARGV[3]
+		local member = ARGV[4]
+	
+		local res = redis.call('set', cacheKey, '', 'nx', 'ex', expireTime)
+	
+		if res then
+			redis.call('zadd', zsetKey, score, member)
+			return "SUCCESS"
+		end
+		return "ERROR"
+		`
+
+		// TODO:
+		expireTime := time.Minute
+
+		res, err := c.redis.Eval(c.ctx, script, []string{c.zsetKey}, cacheKey, expireTime.Seconds(), nextTime.UnixMilli(), val.TaskId).Result()
+
+		valueNum++
+
+		if err == nil && res.(string) == "SUCCESS" {
+			// 设置成功
+			valueNum = 10
+
+			// fmt.Println("计算时间2", val.ExtendData, time.UnixMilli(nextTime.UnixMilli()).Format("2006-01-02 15:04:05"))
+		}
+
+		c.cache.Set(cacheKey, valueNum, expireTime)
 
-		p.ZAdd(ctx, c.zsetKey, &redis.Z{
-			Score:  float64(nextTime.UnixMilli()),
-			Member: val.UniqueKey,
-		})
-		// log.Println("computeTime add", c.zsetKey, val.UniqueKey, nextTime.UnixMilli())
 		return true
 	})
 
-	// 更新缓存
-	b, _ := json.Marshal(execTime)
-	p.Set(ctx, c.nextKey, string(b), 0)
-
-	_, err := p.Exec(ctx)
-	_ = err
-}
-
-// 递归遍历获取执行时间
-func getNextExecTime(beforeTime time.Time, spaceTime time.Duration) time.Time {
-	nowTime := time.Now()
-	if beforeTime.After(nowTime) {
-		return beforeTime
-	}
-	nextTime := beforeTime.Add(spaceTime)
-	if nextTime.Before(nowTime) {
-		nextTime = getNextExecTime(nextTime, spaceTime)
-	}
-	return nextTime
+	// _, err := p.Exec(ctx)
+	// _ = err
 }
 
 // 获取任务
@@ -207,42 +347,109 @@ func (c *Cluster) getTask() {
 // 监听任务
 func (c *Cluster) watch() {
 	// 执行任务
-	for {
-		keys, err := c.redis.BLPop(c.ctx, time.Second*10, c.listKey).Result()
-		if err != nil {
-			fmt.Println("watch err:", err)
-			continue
-		}
-		go doTask(c.ctx, c.redis, keys[1])
-	}
-}
+	go func() {
+		for {
+			keys, err := c.redis.BLPop(c.ctx, time.Second*10, c.listKey).Result()
+			if err != nil {
+				if err != redis.Nil {
+					c.logger.Errorf(c.ctx, "BLPop watch err:%+v", err)
+				}
+				continue
+			}
+			_, ok := clusterWorkerList.Load(keys[1])
+			if !ok {
+				c.logger.Errorf(c.ctx, "watch timer:任务不存在%+v", keys[1])
 
-// 执行任务
-func doTask(ctx context.Context, red *redis.Client, taskId string) {
+				rd := ReJobData{
+					TaskId: keys[1],
+					Times:  1,
+				}
+				rdb, _ := json.Marshal(rd)
 
-	defer func() {
-		if err := recover(); err != nil {
-			fmt.Println("timer:定时器出错", err)
-			log.Println("errStack", string(debug.Stack()))
+				c.redis.SAdd(c.ctx, c.setKey, string(rdb))
+				continue
+			}
+			go c.doTask(c.ctx, keys[1])
 		}
 	}()
 
+	// 处理重入任务
+	go func() {
+		for {
+			res, err := c.redis.SPop(c.ctx, c.setKey).Result()
+			if err != nil {
+				if err == redis.Nil {
+					// 已经是空了就不要浪费资源了
+					time.Sleep(time.Second)
+				} else {
+					c.logger.Errorf(c.ctx, "SPop watch err:%+v", err)
+				}
+				continue
+			}
+
+			var rd ReJobData
+			err = json.Unmarshal([]byte(res), &rd)
+			if err != nil {
+				c.logger.Errorf(c.ctx, "json.Unmarshal err:%+v", err)
+				continue
+			}
+
+			_, ok := clusterWorkerList.Load(rd.TaskId)
+			if !ok {
+				c.logger.Errorf(c.ctx, "watch timer:任务不存在%+v", rd.TaskId)
+
+				if rd.Times >= 3 {
+					// 重试3次还是失败就不执行了
+					continue
+				}
+				rd.Times++
+
+				rdb, _ := json.Marshal(rd)
+
+				c.redis.SAdd(c.ctx, c.setKey, string(rdb))
+				continue
+			}
+			go c.doTask(c.ctx, rd.TaskId)
+		}
+	}()
+
+}
+
+type ReJobData struct {
+	TaskId string
+	Times  int
+}
+
+// 执行任务
+func (c *Cluster) doTask(ctx context.Context, taskId string) {
+
+	ctx, cancel := context.WithTimeout(ctx, c.timeout)
+	defer cancel()
+
 	val, ok := clusterWorkerList.Load(taskId)
 	if !ok {
-		fmt.Println("doTask timer:任务不存在", taskId)
+		c.logger.Errorf(ctx, "doTask timer:任务不存在:%s", taskId)
 		return
 	}
 	t := val.(timerStr)
 
 	// 这里加一个全局锁
-	lock := lockx.NewGlobalLock(ctx, red, taskId)
+	lock := lockx.NewGlobalLock(ctx, c.redis, taskId)
 	tB := lock.Lock()
 	if !tB {
-		fmt.Println("doTask timer:获取锁失败", taskId)
+		c.logger.Errorf(ctx, "doTask timer:获取锁失败:%s", taskId)
 		return
 	}
 	defer lock.Unlock()
 
+	defer func() {
+		if err := recover(); err != nil {
+			c.logger.Errorf(ctx, "timer:回调任务panic err:%+v stack:%s", err, string(debug.Stack()))
+		}
+	}()
+
+	ctx = context.WithValue(ctx, "trace_id", uuid.NewV4().String)
+
 	// 执行任务
-	t.Callback(ctx,t.ExtendData)
+	t.Callback(ctx, t.ExtendData)
 }

cluster_test.go

@@ -1,25 +1,185 @@
-package timerx
+package timerx_test
 
 import (
+	"context"
 	"fmt"
 	"testing"
+	"time"
 
 	"github.com/go-redis/redis/v8"
+	"github.com/yuninks/timerx"
 )
 
-// 示例测试
+func TestCluster_AddEveryMonth(t *testing.T) {
+	ctx := context.Background()
+	redis := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+	})
+	defer redis.Close()
 
-// func exampleDemo(ctx context.Context) bool {
-// 	fmt.Println("fff")
-// 	return false
-// }
+	cluster := timerx.InitCluster(ctx, redis, "test")
 
-// func ExampleB() {
-// 	ctx := context.Background()
-// 	timer.InitSingle(ctx)
-// 	timer.AddToTimer(1, exampleDemo)
-// 	// OutPut:
-// }
+	taskId := "testTask"
+	hour := 2
+	minute := 3
+	second := 4
+	callback := func(ctx context.Context, data interface{}) error {
+		// do something
+		fmt.Println("Task executed:", data)
+		return nil
+	}
+	extendData := "testData"
+
+	err := cluster.EveryMonth(ctx, taskId, 1, hour, minute, second, callback, extendData)
+	if err != nil {
+		t.Errorf("AddEveryMonth failed, err: %v", err)
+	}
+
+	// TODO: verify the job is added to the cluster and can be executed at the specified time
+}
+
+func TestCluster_AddEveryWeek(t *testing.T) {
+	ctx := context.Background()
+	redis := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+	})
+	defer redis.Close()
+
+	cluster := timerx.InitCluster(ctx, redis, "test")
+
+	taskId := "testTask"
+	week := time.Sunday
+	hour := 2
+	minute := 3
+	second := 4
+	callback := func(ctx context.Context, data interface{}) error {
+		// do something
+		fmt.Println("Task executed:", data)
+		return nil
+	}
+	extendData := "testData"
+
+	err := cluster.EveryWeek(ctx, taskId, week, hour, minute, second, callback, extendData)
+	if err != nil {
+		t.Errorf("AddEveryWeek failed, err: %v", err)
+	}
+
+	// TODO: verify the job is added to the cluster and can be executed at the specified time
+}
+
+func TestCluster_AddEveryDay(t *testing.T) {
+	ctx := context.Background()
+	redis := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+	})
+	defer redis.Close()
+
+	cluster := timerx.InitCluster(ctx, redis, "test")
+
+	taskId := "testTask"
+	hour := 2
+	minute := 3
+	second := 4
+	callback := func(ctx context.Context, data interface{}) error {
+		// do something
+		fmt.Println("Task executed:", data)
+		return nil
+	}
+	extendData := "testData"
+
+	err := cluster.EveryDay(ctx, taskId, hour, minute, second, callback, extendData)
+	if err != nil {
+		t.Errorf("AddEveryDay failed, err: %v", err)
+	}
+
+	// TODO: verify the job is added to the cluster and can be executed at the specified time
+}
+
+func TestCluster_AddEveryHour(t *testing.T) {
+	ctx := context.Background()
+	redis := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+	})
+	defer redis.Close()
+
+	cluster := timerx.InitCluster(ctx, redis, "test")
+
+	taskId := "testTask"
+	minute := 3
+	second := 4
+	callback := func(ctx context.Context, data interface{}) error{
+		// do something
+		fmt.Println("Task executed:", data)
+		return nil
+	}
+	extendData := "testData"
+
+	err := cluster.EveryHour(ctx, taskId, minute, second, callback, extendData)
+	if err != nil {
+		t.Errorf("AddEveryHour failed, err: %v", err)
+	}
+
+	// TODO: verify the job is added to the cluster and can be executed at the specified time
+}
+
+func TestCluster_AddEveryMinute(t *testing.T) {
+	ctx := context.Background()
+	redis := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+	})
+	defer redis.Close()
+
+	cluster := timerx.InitCluster(ctx, redis, "test")
+
+	taskId := "testTask"
+	second := 4
+	callback := func(ctx context.Context, data interface{}) error{
+		// do something
+		fmt.Println("Task executed:", data)
+		return nil
+	}
+	extendData := "testData"
+
+	err := cluster.EveryMinute(ctx, taskId, second, callback, extendData)
+	if err != nil {
+		t.Errorf("AddEveryMinute failed, err: %v", err)
+	}
+
+	// TODO: verify the job is added to the cluster and can be executed at the specified time
+}
+
+func TestCluster_Add(t *testing.T) {
+	fmt.Println("66666")
+	ctx := context.Background()
+	fmt.Println("66666")
+	redis := redis.NewClient(&redis.Options{
+		Addr: "localhost:6379",
+	})
+	defer redis.Close()
+
+	t.Log("6666")
+
+	cluster := timerx.InitCluster(ctx, redis, "test")
+
+	taskId := "testTask"
+	dur := time.Second
+	callback := func(ctx context.Context, data interface{}) error {
+		// do something
+		fmt.Println("Task executed:", data)
+		return nil
+	}
+	extendData := "testData"
+
+	err := cluster.EverySpace(ctx, taskId, dur, callback, extendData)
+	if err != nil {
+		t.Errorf("Add failed, err: %v", err)
+	}
+
+	time.Sleep(time.Second * 20)
+
+
+	// TODO: verify the job is added to the cluster and can be executed after the specified duration
+}
 
 func TestMain(m *testing.M) {
 	client := redis.NewClient(&redis.Options{

cmd/main.go

@@ -2,11 +2,12 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"time"
 
-	"code.yun.ink/pkg/timerx"
 	"github.com/go-redis/redis/v8"
+	"github.com/yuninks/timerx"
 )
 
 func main() {
@@ -24,46 +25,100 @@ func main() {
 
 	// re()
 	// d()
-	worker()
+	// cluster()
+	once()
 
+	select {}
+
+}
+
+func once() {
+	client := getRedis()
+	ctx := context.Background()
+	w := OnceWorker{}
+	one := timerx.InitOnce(ctx, client, "test", w)
+
+	d := OnceData{
+		Num: 1,
+	}
+	dy, _ := json.Marshal(d)
+
+	err := one.Create("test", "test", 1*time.Second, dy)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+}
+
+type OnceData struct {
+	Num int
+}
+
+type OnceWorker struct{}
+
+func (l OnceWorker) Worker(ctx context.Context, taskType string, taskId string, attachData []byte) *timerx.OnceWorkerResp {
+	fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
+	fmt.Println(taskId, taskType)
+	fmt.Printf("原来的参数：%s\n", string(attachData))
+
+	d := OnceData{}
+
+	json.Unmarshal(attachData, &d)
+
+	d.Num++
+
+	fmt.Println(d)
+
+	dy, _ := json.Marshal(d)
+
+	return &timerx.OnceWorkerResp{
+		Retry:      true,
+		AttachData: dy,
+		DelayTime:  1 * time.Second,
+	}
+}
+
+func cluster() {
+	client := getRedis()
+	ctx := context.Background()
+	cluster := timerx.InitCluster(ctx, client, "test")
+	err := cluster.EverySpace(ctx, "test_space", 1*time.Second, aa, "这是秒任务")
+	fmt.Println(err)
+	err = cluster.EveryMinute(ctx, "test_min", 15, aa, "这是分钟任务")
+	fmt.Println(err)
+	err = cluster.EveryHour(ctx, "test_hour", 30, 0, aa, "这是小时任务")
+	fmt.Println(err)
+	err = cluster.EveryDay(ctx, "test_day", 11, 0, 0, aa, "这是天任务")
+	fmt.Println(err)
 }
 
 func worker() {
-	client := getRedis()
-	w := timerx.InitOnce(context.Background(), client, &Worker{})
-	w.Add("test", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test2", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test3", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test4", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test5", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
+	// client := getRedis()
+	// w := timerx.InitOnce(context.Background(), client, "test", &OnceWorker{})
+	// w.Save("test", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test2", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test3", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test4", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test5", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
 
 	select {}
 }
 
-type Worker struct{}
-
-func (w *Worker) Worker(uniqueKey string, jobType string, data interface{}) (timerx.WorkerCode, time.Duration) {
-	fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
-	fmt.Println(uniqueKey, jobType)
-	fmt.Println(data)
-	return timerx.WorkerCodeAgain,time.Second
-}
-
 func getRedis() *redis.Client {
 	client := redis.NewClient(&redis.Options{
 		Addr:     "127.0.0.1" + ":" + "6379",
-		Password: "", // no password set
-		DB:       0,  // use default DB
+		Password: "123456", // no password set
+		DB:       0,        // use default DB
 	})
 	if client == nil {
 		panic("redis init error")
@@ -76,21 +131,21 @@ func re() {
 	client := getRedis()
 
 	ctx := context.Background()
-	cl := timerx.InitCluster(ctx, client)
-	cl.Add(ctx, "test1", 1*time.Millisecond, aa, "data")
-	cl.Add(ctx, "test2", 1*time.Millisecond, aa, "data")
-	cl.Add(ctx, "test3", 1*time.Millisecond, aa, "data")
-	cl.Add(ctx, "test4", 1*time.Millisecond, aa, "data")
-	cl.Add(ctx, "test5", 1*time.Millisecond, aa, "data")
-	cl.Add(ctx, "test6", 1*time.Millisecond, aa, "data")
+	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")
 
 	select {}
 }
 
 func aa(ctx context.Context, data interface{}) error {
-	fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
-	fmt.Println(data)
-	time.Sleep(time.Second * 5)
+	fmt.Println("-执行时间:", data, time.Now().Format("2006-01-02 15:04:05"))
+	// fmt.Println(data)
+	// time.Sleep(time.Second * 5)
 	return nil
 }
 

go.mod

@@ -1,11 +1,16 @@
-module code.yun.ink/pkg/timerx
+module github.com/yuninks/timerx
 
 go 1.19
 
-require github.com/go-redis/redis/v8 v8.11.5
+require (
+	github.com/go-redis/redis/v8 v8.11.5
+	github.com/satori/go.uuid v1.2.0
+	github.com/yuninks/cachex v1.0.5
+	github.com/yuninks/lockx v1.0.2
+)
 
 require (
-	code.yun.ink/pkg/lockx v1.0.0 // indirect
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c // indirect
 )

go.sum

@@ -1,5 +1,3 @@
-code.yun.ink/pkg/lockx v1.0.0 h1:xoLyf05PrOAhLID2LbJsEXA8YYURJTK/7spEk/hu/Rs=
-code.yun.ink/pkg/lockx v1.0.0/go.mod h1:0xUU5xD8fui0Kf7g4TnFmaxUDo59CH2WM+sitko2SLc=
 github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
 github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
@@ -7,11 +5,24 @@ github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cu
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
 github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
+github.com/kr/pretty v0.2.1 h1:Fmg33tUaq4/8ym9TJN1x7sLJnHVwhP33CNkpYV/7rwI=
+github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
+github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
+github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
+github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
 github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
+github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
+github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
+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.0.2 h1:p0n791WmsU8D7YF2tQaNLwPE75jdd774unlJZRTNfaw=
+github.com/yuninks/lockx v1.0.2/go.mod h1:J6wvuUELLcMn6FCmiZFt7K5w1QQAh1myL7h3JrZaQiQ=
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
 golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
+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=
 gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=

logger.go

@@ -0,0 +1,25 @@
+package timerx
+
+import (
+	"context"
+	"log"
+)
+
+type Logger interface {
+	Infof(ctx context.Context, format string, v ...interface{})
+	Errorf(ctx context.Context, format string, v ...interface{})
+}
+
+type defaultLogger struct{}
+
+func NewLogger() *defaultLogger {
+	return &defaultLogger{}
+}
+
+func (l *defaultLogger) Infof(ctx context.Context, format string, v ...interface{}) {
+	log.Printf("[INFO] "+format, v...)
+}
+
+func (l *defaultLogger) Errorf(ctx context.Context, format string, v ...interface{}) {
+	log.Printf("[ERROR] "+format, v...)
+}

next_time.go

@@ -0,0 +1,117 @@
+package timerx
+
+import (
+	"errors"
+	"time"
+)
+
+// 计算该任务下次执行时间
+// @param job *JobData 任务数据
+// @return time.Time 下次执行时间
+func GetNextTime(t time.Time, job JobData) (*time.Time, error) {
+
+	var next time.Time
+
+	switch job.JobType {
+	case JobTypeEveryMonth:
+		next = calculateNextMonthTime(t, job)
+	case JobTypeEveryWeek:
+		next = calculateNextWeekTime(t, job)
+	case JobTypeEveryDay:
+		next = calculateNextDayTime(t, job)
+	case JobTypeEveryHour:
+		next = calculateNextHourTime(t, job)
+	case JobTypeEveryMinute:
+		next = calculateNextMinuteTime(t, job)
+	case JobTypeInterval:
+		next = calculateNextInterval(t, job)
+	default:
+		return nil, errors.New("未知的任务类型: " + string(job.JobType))
+	}
+
+	return &next, nil
+}
+
+func calculateNextInterval(t time.Time, job JobData) time.Time {
+	// 从创建的时候开始计算
+	cycle := t.Sub(job.BaseTime).Microseconds() / job.IntervalTime.Microseconds()
+	return job.BaseTime.Add(job.IntervalTime * time.Duration(cycle+1))
+}
+
+func calculateNextMonthTime(t time.Time, job JobData) time.Time {
+	// 判断是否可执行并返回下一个执行时间
+
+	if canRun(t, job) {
+		return time.Date(t.Year(), t.Month(), job.Day, job.Hour, job.Minute, job.Second, 0, t.Location())
+	}
+	// 下一个周期（下个月）
+	return time.Date(t.Year(), t.Month()+1, job.Day, job.Hour, job.Minute, job.Second, 0, t.Location())
+}
+
+func calculateNextWeekTime(t time.Time, job JobData) time.Time {
+	weekday := t.Weekday()
+	days := int(job.Weekday - weekday)
+	if days < 0 {
+		days += 7
+	}
+	// 判断是否可执行并返回下一个执行时间
+	if canRun(t, job) {
+		return time.Date(t.Year(), t.Month(), t.Day(), job.Hour, job.Minute, job.Second, 0, t.Location())
+	}
+	// 下一个周期（下周）
+	return time.Date(t.Year(), t.Month(), t.Day()+days+7, job.Hour, job.Minute, job.Second, 0, t.Location())
+}
+
+func calculateNextDayTime(t time.Time, job JobData) time.Time {
+	// 判断是否可执行并返回下一个执行时间
+	if canRun(t, job) {
+		return time.Date(t.Year(), t.Month(), t.Day(), job.Hour, job.Minute, job.Second, 0, t.Location())
+	}
+	// 下一个周期（明天）
+	return time.Date(t.Year(), t.Month(), t.Day()+1, job.Hour, job.Minute, job.Second, 0, t.Location())
+}
+
+func calculateNextHourTime(t time.Time, job JobData) time.Time {
+	// 判断是否可执行并返回下一个执行时间
+	if canRun(t, job) {
+		return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), job.Minute, job.Second, 0, t.Location())
+	}
+	// 下一个周期（下个小时）
+	return time.Date(t.Year(), t.Month(), t.Day(), t.Hour()+1, job.Minute, job.Second, 0, t.Location())
+}
+
+func calculateNextMinuteTime(t time.Time, job JobData) time.Time {
+	// 判断是否可执行并返回下一个执行时间
+	if canRun(t, job) {
+		return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), job.Second, 0, t.Location())
+	}
+	// 下一个周期（下分钟）
+	return time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute()+1, job.Second, 0, t.Location())
+}
+
+// 检查是否本周期可以运行
+func canRun(t time.Time, job JobData) bool {
+	switch job.JobType {
+	case JobTypeEveryMonth:
+		return t.Day() < job.Day ||
+			(t.Day() == job.Day && t.Hour() < job.Hour) ||
+			(t.Day() == job.Day && t.Hour() == job.Hour && t.Minute() < job.Minute) ||
+			(t.Day() == job.Day && t.Hour() == job.Hour && t.Minute() == job.Minute && t.Second() <= job.Second)
+	case JobTypeEveryWeek:
+		return t.Weekday() < job.Weekday ||
+			(t.Weekday() == job.Weekday && t.Hour() < job.Hour) ||
+			(t.Weekday() == job.Weekday && t.Hour() == job.Hour && t.Minute() < job.Minute) ||
+			(t.Weekday() == job.Weekday && t.Hour() == job.Hour && t.Minute() == job.Minute && t.Second() <= job.Second)
+	case JobTypeEveryDay:
+		return t.Hour() < job.Hour ||
+			(t.Hour() == job.Hour && t.Minute() < job.Minute) ||
+			(t.Hour() == job.Hour && t.Minute() == job.Minute && t.Second() <= job.Second)
+	case JobTypeEveryHour:
+		return t.Minute() < job.Minute ||
+			(t.Minute() == job.Minute && t.Second() <= job.Second)
+	case JobTypeEveryMinute:
+		return t.Second() <= job.Second
+	default:
+		return false
+	}
+}

next_time_test.go

@@ -0,0 +1,108 @@
+package timerx_test
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/yuninks/timerx"
+)
+
+func TestGetNextTime(t *testing.T) {
+	// Test cases
+	tests := []struct {
+		name          string
+		job           timerx.JobData
+		expectedTime  time.Time
+		expectedError error
+	}{
+		{
+			name: "Test JobTypeEveryMonth",
+			job: timerx.JobData{
+				JobType: timerx.JobTypeEveryMonth,
+				Day:     15,
+				Hour:    10,
+				Minute:  0,
+				Second:  0,
+			},
+			expectedTime:  time.Date(2022, 3, 15, 10, 0, 0, 0, time.Local),
+			expectedError: nil,
+		},
+		{
+			name: "Test JobTypeEveryWeek",
+			job: timerx.JobData{
+				JobType: timerx.JobTypeEveryWeek,
+				Weekday: time.Tuesday,
+				Hour:    10,
+				Minute:  0,
+				Second:  0,
+			},
+			expectedTime:  time.Date(2022, 3, 8, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022
+			expectedError: nil,
+		},
+		{
+			name: "Test JobTypeEveryDay",
+			job: timerx.JobData{
+				JobType: timerx.JobTypeEveryDay,
+				Hour:    10,
+				Minute:  0,
+				Second:  0,
+			},
+			expectedTime:  time.Date(2022, 3, 8, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022
+			expectedError: nil,
+		},
+		{
+			name: "Test JobTypeEveryHour",
+			job: timerx.JobData{
+				JobType: timerx.JobTypeEveryHour,
+				Minute:  0,
+				Second:  0,
+			},
+			expectedTime:  time.Date(2022, 3, 7, 11, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
+			expectedError: nil,
+		},
+		{
+			name: "Test JobTypeEveryMinute",
+			job: timerx.JobData{
+				JobType: timerx.JobTypeEveryMinute,
+				Second:  0,
+			},
+			expectedTime:  time.Date(2022, 3, 7, 10, 31, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
+			expectedError: nil,
+		},
+		{
+			name: "Test JobTypeInterval",
+			job: timerx.JobData{
+				JobType:      timerx.JobTypeInterval,
+				IntervalTime: 1 * time.Hour,
+			},
+			expectedTime:  time.Date(2022, 3, 7, 11, 30, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
+			expectedError: nil,
+		},
+		{
+			name: "Test unknown JobType",
+			job: timerx.JobData{
+				JobType: timerx.JobType(100),
+			},
+			expectedTime:  time.Time{},
+			expectedError: errors.New("未知的任务类型: 100"),
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			now := time.Now()
+			// loc := time.FixedZone("CST", 8*3600)
+			nextTime, err := timerx.GetNextTime(now, test.job)
+			if err != nil {
+				if test.expectedError == nil || err.Error() != test.expectedError.Error() {
+					t.Errorf("Expected error: %v, Got error: %v", test.expectedError, err)
+				}
+			} else {
+				if nextTime.IsZero() != (test.expectedTime == time.Time{}) || (nextTime != &test.expectedTime) {
+					t.Errorf("Expected time: %v, Got time: %v", test.expectedTime, nextTime)
+				}
+			}
+		})
+	}
+}

once.go

@@ -4,63 +4,66 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"strings"
 	"sync"
 	"time"
 
 	"github.com/go-redis/redis/v8"
+	uuid "github.com/satori/go.uuid"
 )
 
 // 功能描述
-// 1. 任务全局唯一
-// 2. 任务只执行一次
-// 3. 任务执行失败可以重新放入队列
+// 1. 任务可以多节点发布
+// 2. 每个任务的执行在全局仅会执行一次
+// 3. 任务执行失败支持快捷重新加入队列
 
 // 单次的任务队列
-type worker struct {
+type Once struct {
 	ctx     context.Context
+	logger  Logger
 	zsetKey string
 	listKey string
-	redis   *redis.Client
+	redis   redis.UniversalClient
 	worker  Callback
 }
 
-type WorkerCode int
-
-const (
-	WorkerCodeSuccess WorkerCode = 0  // 处理完成(不需要重入)
-	WorkerCodeAgain   WorkerCode = -1 // 需要继续定时,默认原来的时间
-)
+type OnceWorkerResp struct {
+	Retry      bool // 是否重试 true
+	DelayTime  time.Duration
+	AttachData []byte
+}
 
 // 需要考虑执行失败重新放入队列的情况
 type Callback interface {
 	// 任务执行
-	// uniqueKey: 任务唯一标识
-	// jobType: 任务类型，用于区分任务
-	// data: 任务数据
-	Worker(uniqueKey string, jobType string, data interface{}) (WorkerCode, time.Duration)
+	// @param jobType string 任务类型
+	// @param uniTaskId string 任务唯一标识
+	// @param data interface{} 任务数据
+	// @return WorkerCode 任务执行结果
+	// @return time.Duration 任务执行时间间隔
+	Worker(ctx context.Context, taskType string, taskId string, attachData []byte) *OnceWorkerResp
 }
 
-var wo *worker = nil
+var wo *Once = nil
 var once sync.Once
 
 type extendData struct {
 	Delay time.Duration
-	Data  interface{}
+	Data  []byte
 }
 
 // 初始化
-func InitOnce(ctx context.Context, re *redis.Client, w Callback) *worker {
-
+func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, call Callback, opts ...Option) *Once {
+	op := newOptions(opts...)
 	once.Do(func() {
-		wo = &worker{
+		wo = &Once{
 			ctx:     ctx,
-			zsetKey: "timer:once_zsetkey",
-			listKey: "timer:once_listkey",
+			logger:  op.logger,
+			zsetKey: "timer:once_zsetkey" + keyPrefix,
+			listKey: "timer:once_listkey" + keyPrefix,
 			redis:   re,
-			worker:  w,
+			worker:  call,
 		}
 		go wo.getTask()
 		go wo.watch()
@@ -69,9 +72,13 @@ func InitOnce(ctx context.Context, re *redis.Client, w Callback) *worker {
 	return wo
 }
 
-// 添加任务
+// 添加任务(覆盖)
 // 重复插入就代表覆盖
-func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration, data interface{}) error {
+// @param jobType string 任务类型
+// @param uniTaskId string 任务唯一标识
+// @param delayTime time.Duration 延迟时间
+// @param attachData interface{} 附加数据
+func (w *Once) Save(taskType string, taskId string, delayTime time.Duration, attachData []byte) error {
 	if delayTime.Abs() != delayTime {
 		return fmt.Errorf("时间间隔不能为负数")
 	}
@@ -79,19 +86,21 @@ func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration,
 		return fmt.Errorf("时间间隔不能为0")
 	}
 
-	redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType)
+	redisKey := fmt.Sprintf("%s[:]%s", taskType, taskId)
 
 	ed := extendData{
 		Delay: delayTime,
-		Data:  data,
+		Data:  attachData,
 	}
 	b, _ := json.Marshal(ed)
 
+	// 写入附加数据
 	_, err := w.redis.SetEX(w.ctx, redisKey, b, delayTime+time.Second*5).Result()
 	if err != nil {
 		return err
 	}
 
+	// 写入执行时间
 	_, err = w.redis.ZAdd(w.ctx, w.zsetKey, &redis.Z{
 		Score:  float64(time.Now().Add(delayTime).UnixMilli()),
 		Member: redisKey,
@@ -100,9 +109,38 @@ func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration,
 	return err
 }
 
+// 添加任务(不覆盖)
+func (l *Once) Create(taskType string, taskId string, delayTime time.Duration, attachData []byte) error {
+
+	// 判断有序集合Key是否存在，存在则报错，不存在则写入
+	if l.redis.Exists(l.ctx, l.zsetKey).Val() == 0 {
+		redisKey := fmt.Sprintf("%s[:]%s", taskType, taskId)
+		ed := extendData{
+			Delay: delayTime,
+			Data:  attachData,
+		}
+		b, _ := json.Marshal(ed)
+
+		// 写入附加数据
+		_, err := l.redis.SetEX(l.ctx, redisKey, b, delayTime+time.Second*5).Result()
+		if err != nil {
+			return err
+		}
+		_, err = l.redis.ZAdd(l.ctx, l.zsetKey, &redis.Z{
+			Score:  float64(time.Now().Add(delayTime).UnixMilli()),
+			Member: redisKey,
+		}).Result()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // 删除任务
-func (w *worker) Del(uniqueKey string, jobType string) error {
-	redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType)
+func (w *Once) Delete(taskType string, taskId string) error {
+	redisKey := fmt.Sprintf("%s[:]%s", taskType, taskId)
 
 	w.redis.Del(w.ctx, redisKey).Result()
 
@@ -112,7 +150,12 @@ func (w *worker) Del(uniqueKey string, jobType string) error {
 }
 
 // 获取任务
-func (w *worker) getTask() {
+func (l *Once) Get(taskType string, taskId string) {
+	//
+}
+
+// 获取任务
+func (w *Once) getTask() {
 	timer := time.NewTicker(time.Millisecond * 200)
 	defer timer.Stop()
 
@@ -138,47 +181,54 @@ Loop:
 }
 
 // 监听任务
-func (w *worker) watch() {
+func (w *Once) watch() {
 	for {
 		keys, err := w.redis.BLPop(w.ctx, time.Second*10, w.listKey).Result()
 		if err != nil {
-			fmt.Println("watch err:", err)
+			// fmt.Println("watch err:", err)
 			continue
 		}
 
-		go w.doTask(keys[1])
+		ctx := context.WithValue(w.ctx, "trace_id", uuid.NewV4().String())
+
+		go w.doTask(ctx, keys[1])
 
 	}
 }
 
-func (w *worker) doTask(key string) {
+// 执行任务
+func (l *Once) doTask(ctx context.Context, key string) {
+	fmt.Println("任务时间:", time.Now().Format("2006-01-02 15:04:05"))
 	defer func() {
 		if err := recover(); err != nil {
-			fmt.Println("timer:定时器出错", err)
-			log.Println("errStack", string(debug.Stack()))
+			l.logger.Errorf(ctx, "timer:回调任务panic:%s stack:%s", err, string(debug.Stack()))
 		}
 	}()
 
 	s := strings.Split(key, "[:]")
 
 	// 读取数据
-	str, err := w.redis.Get(w.ctx, key).Result()
+	str, err := l.redis.Get(ctx, key).Result()
 	if err != nil {
-		fmt.Println("execJob err:", err)
+		l.logger.Errorf(ctx, "获取数据失败 err:%s", err)
 		return
 	}
 	ed := extendData{}
 	json.Unmarshal([]byte(str), &ed)
 
-	fmt.Println("开始时间：", time.Now().Format("2006-01-02 15:04:05"))
-	code, t := w.worker.Worker(s[0], s[1], ed.Data)
+	resp := l.worker.Worker(ctx, s[0], s[1], ed.Data)
+	if resp == nil {
+		return
+	}
 
-	if code == WorkerCodeAgain {
+	if resp.Retry {
 		// 重新放入队列
-		fmt.Println("重入时间：", time.Now().Format("2006-01-02 15:04:05"))
-		if t != 0 && t == t.Abs() {
-			ed.Delay = t
+		if resp.DelayTime != 0 && resp.DelayTime == resp.DelayTime.Abs() {
+			ed.Delay = resp.DelayTime
 		}
-		w.Add(s[0], s[1], ed.Delay, ed.Data)
+		ed.Data = resp.AttachData
+		l.logger.Infof(ctx, "任务重新放入队列:%s", key)
+		fmt.Println("重入时间:", time.Now().Format("2006-01-02 15:04:05"))
+		l.Create(s[0], s[1], ed.Delay, ed.Data)
 	}
 }

option.go

@@ -0,0 +1,48 @@
+package timerx
+
+import "time"
+
+type Options struct {
+	logger      Logger
+	location    *time.Location
+	timeout time.Duration
+}
+
+func defaultOptions() Options {
+	return Options{
+		logger:      NewLogger(),
+		location:    time.Local,
+		timeout: time.Hour,
+	}
+}
+
+type Option func(*Options)
+
+func newOptions(opts ...Option) Options {
+	o := defaultOptions()
+	for _, opt := range opts {
+		opt(&o)
+	}
+	return o
+}
+
+// 设置日志
+func SetLogger(log Logger) Option {
+	return func(o *Options) {
+		o.logger = log
+	}
+}
+
+// 设定时区
+func SetTimeZone(zone *time.Location) Option {
+	return func(o *Options) {
+		o.location = zone
+	}
+}
+
+// 设置任务最长执行时间
+func SetTimeout(d time.Duration) Option {
+	return func(o *Options) {
+		o.timeout = d
+	}
+}

readme.md

@@ -1,14 +1,25 @@
-开发目标
+# 功能支持
 
-1. 支持单机定时
-2. 支持集群定时
-3. 支持间隔定时
-4. 支持固定时间
-5. 支持全局唯一
+1. 支持本地任务
+2. 支持集群任务
+3. 支持单次任务
+
+# 功能说明
 
 
-设计思想
-1. 不再单独区分单机还是集群，统一按集群处理，单机只是集群里面只有一个节点
-2. 计算和执行分离，计算只负责计算，执行只负责执行，计算和执行之间通过消息队列进行通信
+
+# 功能实现
+
+1. 集群间任务调度和任务的唯一依赖于redis进行实现
 
 
+# 缺陷
+
+1. 集群部署时，存在新旧的代码混合问题，任务调度可能存在问题（需要根据实际需要进行版本上线/下线操作）
+## 方案一
+1. 启动的时候定时向redis注册任务项
+2. 每次计算执行时间的时候根据注册的任务项进行任务计算
+3. 注册任务项需要有下线机制，避免能运行它的节点下线了它还被执行
+
+
+现在有根据要求根据系统时间整点运行任务的要求，这个比简单的定时重复更复杂，因为不但要按时执行，并且不能重复执行，需要全局记录任务执行的状态，由于任务的间隔时间不确定，这个任务执行状态的保存周期也是有变化的

single.go

@@ -5,32 +5,46 @@ package timerx
 import (
 	"context"
 	"errors"
-	"fmt"
-	"log"
 	"runtime/debug"
 	"sync"
 	"time"
+
+	uuid "github.com/satori/go.uuid"
 )
 
-// 定时器
-// 原理：每毫秒的时间触发
-// 单机版重复时间间隔定时器
+// 简单定时器
+// 1. 这个定时器的作用范围是本机
+// 2. 适用简单的时间间隔定时任务
 
-// uuid -> timerStr
-var timerMap = make(map[string]*timerStr)
-var timerMapMux sync.Mutex
+// 定时器结构体
+var singleWorkerList sync.Map
 
-var timerCount int      // 当前定时数目
-var onceLimit sync.Once // 实现单例
+var singleTimerIndex int // 当前定时数目
 
-type Single struct{}
+var singleOnceLimit sync.Once // 实现单例
+
+type Single struct {
+	ctx      context.Context
+	logger   Logger
+	location *time.Location
+}
 
 var sin *Single = nil
 
+var singleNextTime = time.Now() // 下一次执行的时间
+
 // 定时器类
-func InitSingle(ctx context.Context) *Single {
-	onceLimit.Do(func() {
-		sin = &Single{}
+// @param ctx context.Context 上下文
+// @param opts ...Option 配置项
+func InitSingle(ctx context.Context, opts ...Option) *Single {
+	singleOnceLimit.Do(func() {
+		op := newOptions(opts...)
+
+		sin = &Single{
+			ctx:      ctx,
+			logger:   op.logger,
+			location: op.location,
+		}
 
 		timer := time.NewTicker(time.Millisecond * 200)
 		go func(ctx context.Context) {
@@ -38,108 +52,194 @@ func InitSingle(ctx context.Context) *Single {
 			for {
 				select {
 				case t := <-timer.C:
-					if t.Before(nextTime) {
+					if t.Before(singleNextTime) {
 						// 当前时间小于下次发送时间：跳过
 						continue
 					}
 					// 迭代定时器
-					sin.iterator(ctx, t)
+					sin.iterator(ctx)
 					// fmt.Println("timer: 执行")
 				case <-ctx.Done():
 					// 跳出循环
 					break Loop
 				}
 			}
-			log.Println("timer: initend")
+			sin.logger.Infof(ctx, "timer: initend")
 		}(ctx)
 	})
 
 	return sin
 }
 
+// 每月执行一次
+// @param ctx 上下文
+// @param taskId 任务ID
+// @param day 每月的几号
+// @param hour 小时
+// @param minute 分钟
+// @param second 秒
+// @param callback 回调函数
+// @param extendData 扩展数据
+// @return error
+func (c *Single) AddMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int, error) {
+	nowTime := time.Now()
+
+	jobData := JobData{
+		JobType:    JobTypeEveryMonth,
+		CreateTime: nowTime,
+		Day:        day,
+		Hour:       hour,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
+// 每周执行一次
+// @param ctx context.Context 上下文
+// @param taskId string 任务ID
+// @param week time.Weekday 周
+// @param hour int 小时
+// @param minute int 分钟
+// @param second int 秒
+func (c *Single) AddWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int, error) {
+	nowTime := time.Now()
+
+	jobData := JobData{
+		JobType:    JobTypeEveryWeek,
+		CreateTime: nowTime,
+		Weekday:    week,
+		Hour:       hour,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
+// 每天执行一次
+func (c *Single) AddDay(ctx context.Context, taskId string, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int, error) {
+	nowTime := time.Now()
+
+	jobData := JobData{
+		JobType:    JobTypeEveryDay,
+		CreateTime: nowTime,
+		Hour:       hour,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
+// 每小时执行一次
+func (c *Single) AddHour(ctx context.Context, taskId string, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int, error) {
+	nowTime := time.Now()
+
+	jobData := JobData{
+		JobType:    JobTypeEveryHour,
+		CreateTime: nowTime,
+		Minute:     minute,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
+// 每分钟执行一次
+func (c *Single) AddMinute(ctx context.Context, taskId string, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int, error) {
+	nowTime := time.Now()
+
+	jobData := JobData{
+		JobType:    JobTypeEveryMinute,
+		CreateTime: nowTime,
+		Second:     second,
+	}
+
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
+// 特定时间间隔
+func (c *Single) AddSpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int, error) {
+	nowTime := time.Now()
+
+	if spaceTime < 0 {
+		c.logger.Errorf(ctx, "间隔时间不能小于0")
+		return 0, errors.New("间隔时间不能小于0")
+	}
+
+	jobData := JobData{
+		JobType:      JobTypeInterval,
+		CreateTime:   nowTime,
+		IntervalTime: spaceTime,
+	}
+
+	return c.addJob(ctx, jobData, callback, extendData)
+}
+
 // 间隔定时器
 // @param space 间隔时间
 // @param call 回调函数
 // @param extend 附加参数
 // @return int 定时器索引
 // @return error 错误
-func (s *Single) Add(space time.Duration, call callback, extend interface{}) (int, error) {
-	timerMapMux.Lock()
-	defer timerMapMux.Unlock()
+func (l *Single) addJob(ctx context.Context, jobData JobData, call func(ctx context.Context, extendData interface{}) error, extend interface{}) (int, error) {
+	singleTimerIndex += 1
 
-	if space != space.Abs() {
-		return 0, errors.New("space must be positive")
+	_, err := GetNextTime(time.Now().In(l.location), jobData)
+	if err != nil {
+		l.logger.Errorf(ctx, "获取下次执行时间失败:%s", err.Error())
+		return 0, err
 	}
 
-	timerCount += 1
-
-	nowTime := time.Now()
-
 	t := timerStr{
 		Callback:   call,
-		BeginTime:  nowTime,
-		NextTime:   nowTime, // nowTime.Add(space), // 添加任务的时候就执行一次
-		SpaceTime:  space,
 		CanRunning: make(chan struct{}, 1),
-		UniqueKey:  "",
 		ExtendData: extend,
+		JobData:    &jobData,
 	}
 
-	timerMap[fmt.Sprintf("%d", timerCount)] = &t
+	singleWorkerList.Store(singleTimerIndex, t)
 
-	if t.NextTime.Before(nextTime) {
-		// 本条规则下次需要发送的时间小于系统下次发送时间：替换
-		nextTime = t.NextTime
-	}
-
-	return timerCount, nil
+	return singleTimerIndex, nil
 }
 
 // 删除定时器
-func (s *Single) Del(index string) {
-	timerMapMux.Lock()
-	defer timerMapMux.Unlock()
-	delete(timerMap, index)
+func (s *Single) Del(index int) {
+	singleWorkerList.Delete(index)
 }
 
 // 迭代定时器列表
-func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
-	timerMapMux.Lock()
-	defer timerMapMux.Unlock()
+func (s *Single) iterator(ctx context.Context) {
 
-	// fmt.Println("nowTime:", nowTime.Format("2006-01-02 15:04:05.000"))
+	nowTime := time.Now().In(s.location)
 
 	// 默认5秒后（如果没有值就暂停进来5秒）
 	newNextTime := nowTime.Add(time.Second * 5)
 
 	index := 0
-	for _, v := range timerMap {
+	singleWorkerList.Range(func(k, v interface{}) bool {
 		index++
-		v := v
-		// 判断执行的时机
-		if v.NextTime.Before(nowTime) {
-			// fmt.Println("NextTime", v.NextTime.Format("2006-01-02 15:04:05.000"))
+		timeStr := v.(timerStr)
 
-			v.NextTime = v.NextTime.Add(v.SpaceTime)
-
-			// 判断下次执行时间与当前时间
-			if v.NextTime.Before(nowTime) {
-				v.NextTime = nowTime.Add(v.SpaceTime)
-			}
+		if timeStr.JobData.NextTime.Before(nowTime) || timeStr.JobData.NextTime.Equal(nowTime) {
+			// 可执行
+			nextTime, _ := GetNextTime(nowTime, *timeStr.JobData)
+			timeStr.JobData.NextTime = *nextTime
 
 			if index == 1 {
 				// 循环的第一个需要替换默认值
-				newNextTime = v.NextTime
+				newNextTime = timeStr.JobData.NextTime
 			}
 
-			// 获取最小的
-			if v.NextTime.Before(newNextTime) {
+			if nextTime.Before(newNextTime) {
 				// 本规则下次发送时间小于系统下次需要执行的时间：替换
-				newNextTime = v.NextTime
+				newNextTime = *nextTime
 			}
 
 			// 处理中就跳过本次
-			go func(ctx context.Context, v *timerStr) {
+			go func(ctx context.Context, v timerStr) {
 				select {
 				case v.CanRunning <- struct{}{}:
 					defer func() {
@@ -152,23 +252,27 @@ func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
 						}
 					}()
 					// fmt.Printf("timer: 准备执行 %v %v \n", k, v.Tag)
-					s.doTask(ctx, v.Callback, v.UniqueKey, v.ExtendData)
+					s.doTask(ctx, v.Callback, v.ExtendData)
 				default:
 					// fmt.Printf("timer: 已在执行 %v %v \n", k, v.Tag)
 					return
 				}
-			}(ctx, v)
+			}(ctx, timeStr)
+
 		}
-	}
+
+		return true
+
+	})
 
 	// 实际下次时间小于预期下次时间：替换
-	if nextTime.Before(newNextTime) {
+	if singleNextTime.Before(newNextTime) {
 		// 判断一下避免异常
 		if newNextTime.Before(nowTime) {
 			// 比当前时间小
-			nextTime = nowTime
+			singleNextTime = nowTime
 		} else {
-			nextTime = newNextTime
+			singleNextTime = newNextTime
 		}
 	}
 
@@ -177,12 +281,14 @@ func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
 
 // 定时器操作类
 // 这里不应painc
-func (s *Single) doTask(ctx context.Context, call callback, uniqueKey string, extend interface{}) error {
+func (s *Single) doTask(ctx context.Context, call func(ctx context.Context, extendData interface{}) error, extend interface{}) error {
 	defer func() {
 		if err := recover(); err != nil {
-			fmt.Println("timer:定时器出错", err)
-			log.Println("errStack", string(debug.Stack()))
+			s.logger.Errorf(ctx, "timer:回调任务panic err:%+v stack:%s", err, string(debug.Stack()))
 		}
 	}()
+
+	ctx = context.WithValue(ctx, "trace_id", uuid.NewV4().String)
+
 	return call(ctx, extend)
 }

single_test.go

@@ -1,15 +1,20 @@
 package timerx_test
 
-import "testing"
+import (
+	"fmt"
+	"testing"
+)
 
 // 单元测试
 
 func TestHelloWorld(t *testing.T) {
 	// 日志
-	t.Log("hello world")
+	// t.Log("hello world")
 
-	s := "ddd"
-	t.Logf("Log测试%s", s)
+	fmt.Println("hello world")
+
+	// s := "ddd"
+	// t.Logf("Log测试%s", s)
 	// t.Errorf("ErrorF %s", s)
 
 	// 标记错误(继续运行)

types.go

@@ -6,18 +6,38 @@ import (
 )
 
 type timerStr struct {
-	Callback   callback        // 需要回调的方法
-	CanRunning chan (struct{}) // 是否允许执行
-	BeginTime  time.Time       // 初始化任务的时间
-	NextTime   time.Time       // [删]下一次执行的时间
-	SpaceTime  time.Duration   // 任务间隔时间
-	UniqueKey  string          // 全局唯一键
+	Callback   func(ctx context.Context, extendData interface{}) error        // 需要回调的方法
+	CanRunning chan (struct{}) // 是否允许执行(only single)
+	TaskId     string          // 任务ID 全局唯一键(only cluster)
 	ExtendData interface{}     // 附加参数
+	JobData    *JobData        // 任务时间数据
 }
 
+type JobType string
 
-var nextTime = time.Now() // 下一次执行的时间
+const (
+	JobTypeEveryMonth  JobType = "every_month"  // 每月
+	JobTypeEveryWeek   JobType = "every_week"   // 每周
+	JobTypeEveryDay    JobType = "every_day"    // 每天
+	JobTypeEveryHour   JobType = "every_hour"   // 每小时
+	JobTypeEveryMinute JobType = "every_minute" // 每分钟
+	JobTypeEverySecond JobType = "every_second" // 每秒
+	JobTypeInterval    JobType = "interval"     // 指定时间间隔
+)
 
+type JobData struct {
+	JobType      JobType       // 任务类型
+	NextTime     time.Time     // 下次执行时间
+	BaseTime     time.Time     // 基准时间(间隔的基准时间)
+	CreateTime   time.Time     // 任务创建时间
+	IntervalTime time.Duration // 任务间隔时间
+	Month        time.Month    // 每年的第几个月
+	Weekday      time.Weekday  // 每周的周几
+	Day          int           // 每月的第几天
+	Hour         int           // 每天的第几个小时
+	Minute       int           // 每小时的第几分钟
+	Second       int           // 每分钟的第几秒
+}
 
 // 定义各个回调函数
-type callback func(ctx context.Context, extendData interface{}) error
+// type callback func(ctx context.Context, extendData interface{}) error