优化集群定时器的逻辑

cluster.go

@@ -5,7 +5,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -15,11 +14,14 @@ import (
 )
 
 // 功能描述
+
 // 这是基于Redis的定时任务调度器，能够有效的在服务集群里面调度任务，避免了单点压力过高或单点故障问题
 // 由于所有的服务代码是一致的，也就是一个定时任务将在所有的服务都有注册，具体调度到哪个服务运行看调度结果
 
 // 暂不支持删除定时器，因为这个定时器的设计是基于全局的，如果删除了，那么其他服务就不知道了
 
+// TODO:如果获取到任务不能处理的，应放回队列（因为可能新旧代码同时上线，新的添加了任务处理）
+
 // 单例模式
 var clusterOnceLimit sync.Once
 
@@ -29,23 +31,31 @@ var clusterWorkerList sync.Map
 type Cluster struct {
 	ctx     context.Context
 	redis   *redis.Client
+	logger  Logger
 	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.Client, keyPrefix string, opts ...Option) *Cluster {
+	op := newOptions(opts...)
+
 	clusterOnceLimit.Do(func() {
 		clu = &Cluster{
 			ctx:     ctx,
 			redis:   red,
-			lockKey: "timer:cluster_globalLockKey", // 定时器的全局锁
-			nextKey: "timer:cluster_nextKey",
-			zsetKey: "timer:cluster_zsetKey",
-			listKey: "timer:cluster_listKey",
+			logger:  op.logger,
+			lockKey: keyPrefix + "timer:cluster_globalLockKey", // 定时器的全局锁
+			nextKey: keyPrefix + "timer:cluster_nextKey",       // 下一次
+			zsetKey: keyPrefix + "timer:cluster_zsetKey",       // 有序集合
+			listKey: keyPrefix + "timer:cluster_listKey",       // 列表
+			setKey:  keyPrefix + "timer:cluster_setKey",        // 重入集合
 		}
 
 		// 监听任务
@@ -69,38 +79,108 @@ func InitCluster(ctx context.Context, red *redis.Client) *Cluster {
 	return clu
 }
 
-func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Duration, callback callback, extendData interface{}) error {
-	_, ok := clusterWorkerList.Load(uniqueKey)
+// TODO:指定执行时间
+// 1.每月的1号2点执行（如果当月没有这个号就不执行）
+// 2.每周的周一2点执行
+// 3.每天的2点执行
+// 4.每小时的2分执行
+// 5.每分钟的2秒执行
+func (c *Cluster) AddEveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback callback, extendData interface{}) error {
+	nowTime := time.Now()
+	// 计算下一次执行的时间
+	nextTime := time.Date(nowTime.Year(), nowTime.Month(), day, hour, minute, second, 0, nowTime.Location())
+	if nextTime.Before(nowTime) {
+		nextTime = nextTime.AddDate(0, 1, 0)
+	}
+	return c.addJob(ctx, taskId, nextTime, time.Hour*24*30, callback, extendData, JobTypeEveryMonth, &JobData{Day: &day, Hour: &hour, Minute: &minute, Second: &second})
+}
+
+func (c *Cluster) AddEveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback callback, extendData interface{}) error {
+	nowTime := time.Now()
+	// 计算下一次执行的时间
+	nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), hour, minute, second, 0, nowTime.Location())
+	for nextTime.Weekday() != week {
+		nextTime = nextTime.AddDate(0, 0, 1)
+	}
+	if nextTime.Before(nowTime) {
+		nextTime = nextTime.AddDate(0, 0, 7)
+	}
+	return c.addJob(ctx, taskId, nextTime, time.Hour*24*7, callback, extendData, JobTypeInterval, nil)
+}
+
+func (c *Cluster) AddEveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback callback, extendData interface{}) error {
+	nowTime := time.Now()
+	// 计算下一次执行的时间
+	nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), hour, minute, second, 0, nowTime.Location())
+	if nextTime.Before(nowTime) {
+		nextTime = nextTime.AddDate(0, 0, 1)
+	}
+	return c.addJob(ctx, taskId, nextTime, time.Hour*24, callback, extendData, JobTypeInterval, nil)
+}
+
+func (c *Cluster) AddEveryHour(ctx context.Context, taskId string, minute int, second int, callback callback, extendData interface{}) error {
+	nowTime := time.Now()
+	// 计算下一次执行的时间
+	nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), nowTime.Hour(), minute, second, 0, nowTime.Location())
+	if nextTime.Before(nowTime) {
+		nextTime = nextTime.Add(time.Hour)
+	}
+	return c.addJob(ctx, taskId, nextTime, time.Hour, callback, extendData, JobTypeInterval, nil)
+}
+
+func (c *Cluster) AddEveryMinute(ctx context.Context, taskId string, second int, callback callback, extendData interface{}) error {
+	nowTime := time.Now()
+	// 计算下一次执行的时间
+	nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), nowTime.Hour(), nowTime.Minute(), second, 0, nowTime.Location())
+	if nextTime.Before(nowTime) {
+		nextTime = nextTime.Add(time.Minute)
+	}
+	return c.addJob(ctx, taskId, nextTime, time.Minute, callback, extendData, JobTypeInterval, nil)
+}
+
+func (c *Cluster) Add(ctx context.Context, taskId string, spaceTime time.Duration, callback callback, extendData interface{}) error {
+	return c.addJob(ctx, taskId, time.Now(), spaceTime, callback, extendData, JobTypeInterval, nil)
+}
+
+// 指定时间间隔
+// TODO:
+// 1.不同服务定的时间间隔不一致问题
+// 2.后起的服务计算了时间覆盖前面原有的时间问题
+func (c *Cluster) addJob(ctx context.Context, taskId string, beginTime time.Time, spaceTime time.Duration, callback callback, extendData interface{}, jobType JobType, jobData *JobData) error {
+	_, ok := clusterWorkerList.Load(taskId)
 	if ok {
+		c.logger.Errorf(ctx, "key已存在:%s", taskId)
 		return errors.New("key已存在")
 	}
 
 	if spaceTime != spaceTime.Abs() {
+		c.logger.Errorf(ctx, "时间间隔不能为负数:%s", taskId)
 		return errors.New("时间间隔不能为负数")
 	}
 
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
-	lock := lockx.NewGlobalLock(ctx, c.redis, uniqueKey)
+	lock := lockx.NewGlobalLock(ctx, c.redis, taskId)
 	tB := lock.Try(10)
 	if !tB {
+		c.logger.Errorf(ctx, "添加失败:%s", taskId)
 		return errors.New("添加失败")
 	}
 	defer lock.Unlock()
 
-	nowTime := time.Now()
-
 	t := timerStr{
-		BeginTime:  nowTime,
-		NextTime:   nowTime,
+		BeginTime:  beginTime,
+		NextTime:   beginTime,
 		SpaceTime:  spaceTime,
 		Callback:   callback,
 		ExtendData: extendData,
-		UniqueKey:  uniqueKey,
+		TaskId:     taskId,
+		JobType:    jobType,
+		JobData:    jobData,
 	}
 
-	clusterWorkerList.Store(uniqueKey, t)
+	clusterWorkerList.Store(taskId, t)
 
 	cacheStr, _ := c.redis.Get(ctx, c.nextKey).Result()
 	execTime := make(map[string]time.Time)
@@ -110,9 +190,9 @@ func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Dura
 
 	p.ZAdd(ctx, c.zsetKey, &redis.Z{
 		Score:  float64(nextTime.UnixMilli()),
-		Member: uniqueKey,
+		Member: taskId,
 	})
-	execTime[uniqueKey] = nextTime
+	execTime[taskId] = nextTime
 	n, _ := json.Marshal(execTime)
 	// fmt.Println("execTime:", execTime, string(n))
 	p.Set(ctx, c.nextKey, string(n), 0)
@@ -125,6 +205,7 @@ func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Dura
 }
 
 // 计算下一次执行的时间
+// TODO:注册的任务需放在Redis集中存储，因为本地的话，如果有多个服务，那么就会出现不一致的情况。但是要注意服务如何进行下线，由于是主动上报的，需要有一个机制进行删除过期的任务（添加任务&定时器轮训注册）
 func (c *Cluster) getNextTime() {
 
 	// log.Println("begin computer")
@@ -153,18 +234,18 @@ func (c *Cluster) getNextTime() {
 
 	clusterWorkerList.Range(func(key, value interface{}) bool {
 		val := value.(timerStr)
-		beforeTime := execTime[val.UniqueKey]
+		beforeTime := execTime[val.TaskId]
 		if beforeTime.After(nowTime) {
 			return true
 		}
-		nextTime := getNextExecTime(beforeTime, val.SpaceTime)
-		execTime[val.UniqueKey] = nextTime
+		nextTime := getNextExecTime(val)
+		execTime[val.TaskId] = nextTime
 
 		p.ZAdd(ctx, c.zsetKey, &redis.Z{
 			Score:  float64(nextTime.UnixMilli()),
-			Member: val.UniqueKey,
+			Member: val.TaskId,
 		})
-		// log.Println("computeTime add", c.zsetKey, val.UniqueKey, nextTime.UnixMilli())
+		// log.Println("computeTime add", c.zsetKey, val.taskId, nextTime.UnixMilli())
 		return true
 	})
 
@@ -177,14 +258,16 @@ func (c *Cluster) getNextTime() {
 }
 
 // 递归遍历获取执行时间
-func getNextExecTime(beforeTime time.Time, spaceTime time.Duration) time.Time {
+// TODO:需要根据不同的任务类型计算下次定时时间
+func getNextExecTime(ts timerStr) time.Time {
 	nowTime := time.Now()
-	if beforeTime.After(nowTime) {
-		return beforeTime
+	if ts.NextTime.After(nowTime) {
+		return ts.NextTime
 	}
-	nextTime := beforeTime.Add(spaceTime)
+	nextTime := ts.NextTime.Add(ts.SpaceTime)
+	ts.NextTime = nextTime
 	if nextTime.Before(nowTime) {
-		nextTime = getNextExecTime(nextTime, spaceTime)
+		nextTime = getNextExecTime(ts)
 	}
 	return nextTime
 }
@@ -207,29 +290,59 @@ func (c *Cluster) getTask() {
 // 监听任务
 func (c *Cluster) watch() {
 	// 执行任务
+	go func() {
 		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])
+			_, ok := clusterWorkerList.Load(keys[1])
+			if !ok {
+				fmt.Println("watch timer:任务不存在", keys[1])
+				c.redis.SAdd(c.ctx, c.setKey, keys[1])
+				continue
 			}
+			go c.doTask(c.ctx, c.redis, keys[1])
+		}
+	}()
+
+	go func() {
+		for {
+			taskId, err := c.redis.SPop(c.ctx, c.setKey).Result()
+
+			if err != nil {
+				fmt.Println("watch err:", err)
+				if err == redis.Nil {
+					// 已经是空了就不要浪费资源了
+					time.Sleep(time.Second)
+				}
+				continue
+			}
+			_, ok := clusterWorkerList.Load(taskId)
+			if !ok {
+				fmt.Println("watch timer:任务不存在", taskId)
+				c.redis.SAdd(c.ctx, c.setKey, taskId)
+				continue
+			}
+			go c.doTask(c.ctx, c.redis, taskId)
+		}
+	}()
+
 }
 
 // 执行任务
-func doTask(ctx context.Context, red *redis.Client, taskId string) {
+func (c *Cluster) doTask(ctx context.Context, red *redis.Client, taskId string) {
 
 	defer func() {
 		if err := recover(); err != nil {
-			fmt.Println("timer:定时器出错", err)
-			log.Println("errStack", string(debug.Stack()))
+			c.logger.Errorf(ctx, "timer:定时器出错 err:%+v stack:%s", err, string(debug.Stack()))
 		}
 	}()
 
 	val, ok := clusterWorkerList.Load(taskId)
 	if !ok {
-		fmt.Println("doTask timer:任务不存在", taskId)
+		c.logger.Errorf(ctx, "doTask timer:任务不存在", taskId)
 		return
 	}
 	t := val.(timerStr)
@@ -238,7 +351,7 @@ func doTask(ctx context.Context, red *redis.Client, taskId string) {
 	lock := lockx.NewGlobalLock(ctx, red, taskId)
 	tB := lock.Lock()
 	if !tB {
-		fmt.Println("doTask timer:获取锁失败", taskId)
+		c.logger.Errorf(ctx, "doTask timer:获取锁失败", taskId)
 		return
 	}
 	defer lock.Unlock()

cmd/main.go

@@ -76,7 +76,7 @@ func re() {
 	client := getRedis()
 
 	ctx := context.Background()
-	cl := timerx.InitCluster(ctx, client)
+	cl := timerx.InitCluster(ctx, client, "kkkk")
 	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")

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...)
+}

option.go

@@ -0,0 +1,27 @@
+package timerx
+
+type Options struct {
+	logger Logger
+}
+
+func defaultOptions() Options {
+	return Options{
+		logger: NewLogger(),
+	}
+}
+
+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
+	}
+}

single.go

@@ -80,7 +80,6 @@ func (s *Single) Add(space time.Duration, call callback, extend interface{}) (in
 		NextTime:   nowTime, // nowTime.Add(space), // 添加任务的时候就执行一次
 		SpaceTime:  space,
 		CanRunning: make(chan struct{}, 1),
-		UniqueKey:  "",
 		ExtendData: extend,
 	}
 
@@ -151,7 +150,7 @@ 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
@@ -176,7 +175,7 @@ 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 callback, extend interface{}) error {
 	defer func() {
 		if err := recover(); err != nil {
 			fmt.Println("timer:定时器出错", err)

types.go

@@ -11,13 +11,34 @@ type timerStr struct {
 	BeginTime  time.Time       // 初始化任务的时间
 	NextTime   time.Time       // [删]下一次执行的时间
 	SpaceTime  time.Duration   // 任务间隔时间
-	UniqueKey  string          // 全局唯一键
+	TaskId     string          // 任务ID 全局唯一键
 	ExtendData interface{}     // 附加参数
+	JobType    JobType         // 任务类型
+	JobData    *JobData        // 任务时间数据
 }
 
+type JobType string
+
+const (
+	JobTypeEveryDay    JobType = "every_day"
+	JobTypeEveryHour   JobType = "every_hour"
+	JobTypeEveryMinute JobType = "every_minute"
+	JobTypeEverySecond JobType = "every_second"
+	JobTypeEveryMonth  JobType = "every_month"
+	// 根据间隔时间执行
+	JobTypeInterval JobType = "interval"
+)
+
+type JobData struct {
+	Month   *time.Month   // 每年的第几个月
+	Weekday *time.Weekday // 每周的周几
+	Day     *int          // 每月的第几天
+	Hour    *int          // 每天的第几个小时
+	Minute  *int          // 每小时的第几分钟
+	Second  *int          // 每分钟的第几秒
+}
 
 var nextTime = time.Now() // 下一次执行的时间
 
-
 // 定义各个回调函数
 type callback func(ctx context.Context, extendData interface{}) error