一些优化

cluster.go

@@ -48,11 +48,11 @@ func InitCluster(ctx context.Context, red *redis.Client, keyPrefix string, opts
 			ctx:     ctx,
 			redis:   red,
 			logger:  op.logger,
-			lockKey: keyPrefix + "timer:cluster_globalLockKey", // 定时器的全局锁
+			lockKey: "timer:cluster_globalLockKey" + keyPrefix, // 定时器的全局锁
-			nextKey: keyPrefix + "timer:cluster_nextKey",       // 下一次
+			nextKey: "timer:cluster_nextKey" + keyPrefix,       // 下一次
-			zsetKey: keyPrefix + "timer:cluster_zsetKey",       // 有序集合
+			zsetKey: "timer:cluster_zsetKey" + keyPrefix,       // 有序集合
-			listKey: keyPrefix + "timer:cluster_listKey",       // 列表
+			listKey: "timer:cluster_listKey" + keyPrefix,       // 列表
-			setKey:  keyPrefix + "timer:cluster_setKey",        // 重入集合
+			setKey:  "timer:cluster_setKey" + keyPrefix,        // 重入集合
 		}
 
 		// 监听任务
@@ -228,6 +228,7 @@ func (c *Cluster) getNextTime() {
 
 	nowTime := time.Now()
 
+	// 根据内部注册的任务列表计算下一次执行的时间
 	clusterWorkerList.Range(func(key, value interface{}) bool {
 		val := value.(timerStr)
 		beforeTime := execTime[val.TaskId]
@@ -262,6 +263,8 @@ func getNextExecTime(ts timerStr) time.Time {
 	}
 	nextTime := ts.NextTime.Add(ts.SpaceTime)
 	ts.NextTime = nextTime
+
+	// 递归计算直到拿到下一次执行的时间
 	if nextTime.Before(nowTime) {
 		nextTime = getNextExecTime(ts)
 	}
@@ -305,6 +308,7 @@ func (c *Cluster) watch() {
 		}
 	}()
 
+	// 处理重入任务
 	go func() {
 		for {
 			taskId, err := c.redis.SPop(c.ctx, c.setKey).Result()

cmd/main.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"code.yun.ink/pkg/timerx"
+	"github.com/yuninks/timerx"
 	"github.com/go-redis/redis/v8"
 )
 

once.go

@@ -19,8 +19,9 @@ import (
 // 3. 任务执行失败支持快捷重新加入队列
 
 // 单次的任务队列
-type worker struct {
+type Once struct {
 	ctx     context.Context
+	logger  Logger
 	zsetKey string
 	listKey string
 	redis   *redis.Client
@@ -43,7 +44,7 @@ type Callback interface {
 	Worker(jobType string, uniqueKey string, data interface{}) (WorkerCode, time.Duration)
 }
 
-var wo *worker = nil
+var wo *Once = nil
 var once sync.Once
 
 type extendData struct {
@@ -52,15 +53,16 @@ type extendData struct {
 }
 
 // 初始化
-func InitOnce(ctx context.Context, re *redis.Client, jobGlobalName string, jobCallback Callback) *worker {
+func InitOnce(ctx context.Context, re *redis.Client, keyPrefix string, call Callback, opts ...Option) *Once {
-
+	op := newOptions(opts...)
 	once.Do(func() {
-		wo = &worker{
+		wo = &Once{
 			ctx:     ctx,
-			zsetKey: "timer:once_zsetkey" + jobGlobalName,
+			logger:  op.logger,
-			listKey: "timer:once_listkey" + jobGlobalName,
+			zsetKey: "timer:once_zsetkey" + keyPrefix,
+			listKey: "timer:once_listkey" + keyPrefix,
 			redis:   re,
-			worker:  jobCallback,
+			worker:  call,
 		}
 		go wo.getTask()
 		go wo.watch()
@@ -71,7 +73,7 @@ func InitOnce(ctx context.Context, re *redis.Client, jobGlobalName string, jobCa
 
 // 添加任务
 // 重复插入就代表覆盖
-func (w *worker) Add(jobType string, uniqueKey string, delayTime time.Duration, data interface{}) error {
+func (w *Once) Add(jobType string, uniqueKey string, delayTime time.Duration, data interface{}) error {
 	if delayTime.Abs() != delayTime {
 		return fmt.Errorf("时间间隔不能为负数")
 	}
@@ -101,7 +103,7 @@ func (w *worker) Add(jobType string, uniqueKey string, delayTime time.Duration,
 }
 
 // 删除任务
-func (w *worker) Del(jobType string, uniqueKey string) error {
+func (w *Once) Del(jobType string, uniqueKey string) error {
 	redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
 
 	w.redis.Del(w.ctx, redisKey).Result()
@@ -112,7 +114,7 @@ func (w *worker) Del(jobType string, uniqueKey string) error {
 }
 
 // 获取任务
-func (w *worker) getTask() {
+func (w *Once) getTask() {
 	timer := time.NewTicker(time.Millisecond * 200)
 	defer timer.Stop()
 
@@ -138,7 +140,7 @@ 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 {
@@ -151,7 +153,8 @@ func (w *worker) watch() {
 	}
 }
 
-func (w *worker) doTask(key string) {
+// 执行任务
+func (w *Once) doTask(key string) {
 	defer func() {
 		if err := recover(); err != nil {
 			fmt.Println("timer:定时器出错", err)

readme.md

@@ -1,14 +1,19 @@
-开发目标
+# 功能支持
 
-1. 支持单机定时
+1. 支持本地任务
-2. 支持集群定时
+2. 支持集群任务
-3. 支持间隔定时
+3. 支持单次任务
-4. 支持固定时间
+
-5. 支持全局唯一
+# 功能说明
 
 
-设计思想
+
-1. 不再单独区分单机还是集群，统一按集群处理，单机只是集群里面只有一个节点
+# 功能实现
-2. 计算和执行分离，计算只负责计算，执行只负责执行，计算和执行之间通过消息队列进行通信
+
+1. 集群间任务调度和任务的唯一依赖于redis进行实现
 
 
+# 缺陷
+
+1. 集群部署时，存在新旧的代码混合问题，任务调度可能存在问题（需要根据实际需要进行版本上线/下线操作）
+

single.go

@@ -11,8 +11,9 @@ import (
 	"time"
 )
 
-// 定时器
+// 简单定时器
 // 1. 这个定时器的作用范围是本机
+// 2. 适用简单的时间间隔定时任务
 
 // uuid -> timerStr
 var timerMap = make(map[string]*timerStr)