修改mod

.gitlab-ci.yml

@@ -1,26 +1,25 @@
-# You can override the included template(s) by including variable overrides
-# SAST customization: https://docs.gitlab.com/ee/user/application_security/sast/#customizing-the-sast-settings
-# Secret Detection customization: https://docs.gitlab.com/ee/user/application_security/secret_detection/#customizing-settings
-# Dependency Scanning customization: https://docs.gitlab.com/ee/user/application_security/dependency_scanning/#customizing-the-dependency-scanning-settings
-# Container Scanning customization: https://docs.gitlab.com/ee/user/application_security/container_scanning/#customizing-the-container-scanning-settings
-# Note that environment variables can be set in several places
-# See https://docs.gitlab.com/ee/ci/variables/#cicd-variable-precedence
 stages:
-- build
-- test
-- deploy
-- review
-- dast
-- staging
-- canary
-- production
-- incremental rollout 10%
-- incremental rollout 25%
-- incremental rollout 50%
-- incremental rollout 100%
-- performance
-- cleanup
-sast:
-  stage: test
-include:
-- template: Auto-DevOps.gitlab-ci.yml
+  - deploy
+
+before_script:
+  - 'which ssh-agent || ( apt-get update -y && apt-get install openssh-client -y )'
+  - eval $(ssh-agent -s)
+  - echo "$SSH_PRIVATE_KEY" | tr -d '\r' | ssh-add -
+  - mkdir -p ~/.ssh
+  - chmod 700 ~/.ssh
+  - '[[ -f /.dockerenv ]] && echo -e "Host *\n\tStrictHostKeyChecking no\n\n" > ~/.ssh/config'
+
+deploy_job:
+  stage: deploy
+  script:
+    - git remote remove github || true
+    - git remote add github git@github.com:yun-ink/timerx.git
+    - git remote -v
+    - git checkout master
+    - git fsck --full
+    - git prune
+    - git gc --prune=now --aggressive
+    - git push -u github master -f
+  only:
+    - master
+    - tags

cluster.go

@@ -1,45 +1,61 @@
-package timer
+package timerx
 
 import (
 	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"sync"
 	"time"
 
-	"code.yun.ink/open/timer/lockx"
+	"code.yun.ink/pkg/lockx"
 	"github.com/go-redis/redis/v8"
 )
 
+// 功能描述
+
+// 这是基于Redis的定时任务调度器，能够有效的在服务集群里面调度任务，避免了单点压力过高或单点故障问题
+// 由于所有的服务代码是一致的，也就是一个定时任务将在所有的服务都有注册，具体调度到哪个服务运行看调度结果
+
+// 暂不支持删除定时器，因为这个定时器的设计是基于全局的，如果删除了，那么其他服务就不知道了
+
+// TODO:如果获取到任务不能处理的，应放回队列（因为可能新旧代码同时上线，新的添加了任务处理）
+
 // 单例模式
 var clusterOnceLimit sync.Once
 
 // 已注册的任务列表
 var clusterWorkerList sync.Map
 
-type cluster struct {
+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
+var clu *Cluster = nil
+
+// 初始化定时器
+// 全局只需要初始化一次
+func InitCluster(ctx context.Context, red *redis.Client, keyPrefix string, opts ...Option) *Cluster {
+	op := newOptions(opts...)
 
-func InitCluster(ctx context.Context, red *redis.Client) *cluster {
 	clusterOnceLimit.Do(func() {
-		clu = &cluster{
+		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",        // 重入集合
 		}
 
 		// 监听任务
@@ -63,38 +79,108 @@ func InitCluster(ctx context.Context, red *redis.Client) *cluster {
 	return clu
 }
 
-func (c *cluster) AddTimer(ctx context.Context, uniqueKey string, spaceTime time.Duration, callback callback, extend ExtendParams) 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,
-		SpaceTime: spaceTime,
-		Callback:  callback,
-		Extend:    extend,
-		UniqueKey: uniqueKey,
+		BeginTime:  beginTime,
+		NextTime:   beginTime,
+		SpaceTime:  spaceTime,
+		Callback:   callback,
+		ExtendData: extendData,
+		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)
@@ -104,9 +190,9 @@ func (c *cluster) AddTimer(ctx context.Context, uniqueKey string, spaceTime time
 
 	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)
@@ -119,7 +205,8 @@ func (c *cluster) AddTimer(ctx context.Context, uniqueKey string, spaceTime time
 }
 
 // 计算下一次执行的时间
-func (c *cluster) getNextTime() {
+// TODO:注册的任务需放在Redis集中存储，因为本地的话，如果有多个服务，那么就会出现不一致的情况。但是要注意服务如何进行下线，由于是主动上报的，需要有一个机制进行删除过期的任务（添加任务&定时器轮训注册）
+func (c *Cluster) getNextTime() {
 
 	// log.Println("begin computer")
 	ctx, cancel := context.WithCancel(c.ctx)
@@ -147,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
 	})
 
@@ -171,44 +258,23 @@ 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
 }
 
 // 获取任务
-func (c *cluster) getTask() {
+func (c *Cluster) getTask() {
 	// 定时去Redis获取任务
-	// zb := redis.ZRangeBy{
-	// 	Min: "0",
-	// 	Max: fmt.Sprintf("%+v", time.Now().UnixMilli()),
-	// }
-
-	// taskList, _ := c.redis.ZRangeByScore(c.ctx, c.zsetKey, &zb).Result()
-
-	// if len(taskList) == 0 {
-	// 	return
-	// }
-
-	// p := c.redis.Pipeline()
-
-	// for _, val := range taskList {
-	// 	// 添加到可执行队列
-	// 	p.LPush(c.ctx, c.listKey, val)
-	// 	// 删除有序集合
-	// 	p.ZRem(c.ctx, c.zsetKey, val)
-	// }
-	// _, err := p.Exec(c.ctx)
-	// // fmt.Println(err)
-	// _ = err
-
 	script := `
 	local token = redis.call('zrangebyscore',KEYS[1],ARGV[1],ARGV[2])
 	for i,v in ipairs(token) do
@@ -222,33 +288,61 @@ func (c *cluster) getTask() {
 }
 
 // 监听任务
-func (c *cluster) watch() {
+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 func() {
+		for {
+			keys, err := c.redis.BLPop(c.ctx, time.Second*10, c.listKey).Result()
+			if err != nil {
+				fmt.Println("watch err:", err)
+				continue
+			}
+			_, 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 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) {
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
+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)
@@ -257,13 +351,11 @@ 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()
 
-	ctx = context.WithValue(ctx, extendParamKey, t.Extend)
-
 	// 执行任务
-	t.Callback(ctx)
+	t.Callback(ctx, t.ExtendData)
 }

cluster_test.go

@@ -1,4 +1,4 @@
-package timer
+package timerx
 
 import (
 	"fmt"

cmd/main.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"code.yun.ink/open/timer"
+	"code.yun.ink/pkg/timerx"
 	"github.com/go-redis/redis/v8"
 )
 
@@ -30,7 +30,7 @@ func main() {
 
 func worker() {
 	client := getRedis()
-	w := timer.InitWorker(context.Background(), client, &Worker{})
+	w := timerx.InitOnce(context.Background(), client, "test", &Worker{})
 	w.Add("test", "test", 1*time.Second, map[string]interface{}{
 		"test": "test",
 	})
@@ -52,11 +52,11 @@ func worker() {
 
 type Worker struct{}
 
-func (w *Worker) Worker(uniqueKey string, jobType string,data map[string]interface{}) timer.WorkerCode {
+func (w *Worker) Worker(jobType string, uniqueKey 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 timer.WorkerCodeAgain
+	return timerx.WorkerCodeAgain, time.Second
 }
 
 func getRedis() *redis.Client {
@@ -76,63 +76,22 @@ func re() {
 	client := getRedis()
 
 	ctx := context.Background()
-	cl := timer.InitCluster(ctx, client)
-	cl.AddTimer(ctx, "test1", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text1",
-		},
-	})
-	cl.AddTimer(ctx, "test2", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text2",
-		},
-	})
-	cl.AddTimer(ctx, "test3", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text3",
-		},
-	})
-	cl.AddTimer(ctx, "test4", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text4",
-		},
-	})
-	cl.AddTimer(ctx, "test5", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text5",
-		},
-	})
-	cl.AddTimer(ctx, "test6", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text6",
-		},
-	})
-	cl.AddTimer(ctx, "test7", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text7",
-		},
-	})
-	cl.AddTimer(ctx, "test8", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text8",
-		},
-	})
-	cl.AddTimer(ctx, "test9", 1*time.Millisecond, aa, timer.ExtendParams{
-		Params: map[string]interface{}{
-			"test": "text9",
-		},
-	})
+	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")
+	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")
 
 	select {}
 }
 
-func aa(ctx context.Context) bool {
-	// fmt.Println(time.Now().Format(time.RFC3339))
-	// fmt.Println("gggggggggggggggggggggggggggg")
-	a, err := timer.GetExtendParams(ctx)
-	fmt.Printf("%+v %+v \n\n", a, err)
+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)
-	return true
+	return nil
 }
 
 func d() {

go.mod

@@ -1,8 +1,11 @@
-module code.yun.ink/open/timer
+module code.yun.ink/pkg/timerx
 
 go 1.19
 
-require github.com/go-redis/redis/v8 v8.11.5
+require (
+	code.yun.ink/pkg/lockx v1.0.0
+	github.com/go-redis/redis/v8 v8.11.5
+)
 
 require (
 	github.com/cespare/xxhash/v2 v2.1.2 // indirect

go.sum

@@ -1,3 +1,6 @@
+code.yun.ink/open/timer v1.0.1 h1:ZWecU5K0rFB15p8DZubozTEwo1vrO4mUCRwEoD1tbEQ=
+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=

lockx/lockx.go

@@ -1,120 +0,0 @@
-package lockx
-
-import (
-	"context"
-	"fmt"
-	"log"
-	"time"
-
-	"github.com/go-redis/redis/v8"
-)
-
-// 全局锁
-type globalLock struct {
-	redis     *redis.Client
-	ctx       context.Context
-	cancel    context.CancelFunc
-	uniqueKey string
-	value     string
-}
-
-func NewGlobalLock(ctx context.Context, red *redis.Client, uniqueKey string) *globalLock {
-	ctx, cancel := context.WithTimeout(ctx, time.Second*30)
-	return &globalLock{
-		redis:     red,
-		ctx:       ctx,
-		cancel:    cancel,
-		uniqueKey: uniqueKey,
-		value:     fmt.Sprintf("%d", time.Now().UnixNano()),
-	}
-}
-
-// 获取锁
-func (g *globalLock) Lock() bool {
-
-	script := `
-	local token = redis.call('get',KEYS[1])
-	if token == false
-	then
-		return redis.call('set',KEYS[1],ARGV[1],'EX',ARGV[2])
-	end
-		return 'ERROR'
-	`
-
-	resp, err := g.redis.Eval(g.ctx, script, []string{g.uniqueKey}, g.value, 5).Result()
-	if resp != "OK" {
-		_ = err
-		log.Println("globalLock Lock", resp, err, g.uniqueKey, g.value)
-	}
-	if resp == "OK" {
-		g.refresh()
-		return true
-	}
-	return false
-}
-
-// 尝试获取锁
-func (g *globalLock) Try(limitTimes int) bool {
-	for i := 0; i < limitTimes; i++ {
-		if g.Lock() {
-			return true
-		}
-		time.Sleep(time.Millisecond * 100)
-	}
-	return false
-}
-
-// 删除锁
-func (g *globalLock) Unlock() bool {
-
-	script := `
-	local token = redis.call('get',KEYS[1])
-	if token == ARGV[1]
-	then
-		redis.call('del',KEYS[1])
-		return 'OK'
-	end
-		return 'ERROR'
-	`
-
-	resp, err := g.redis.Eval(g.ctx, script, []string{g.uniqueKey}, g.value).Result()
-	if resp != "OK" {
-		log.Println("globalLock Unlock", resp, err, g.uniqueKey, g.value)
-	}
-	g.cancel()
-	return false
-}
-
-// 刷新锁
-func (g *globalLock) refresh() {
-	go func() {
-		t := time.NewTicker(time.Second)
-		for {
-			select {
-			case <-t.C:
-				g.refreshExec()
-			case <-g.ctx.Done():
-				t.Stop()
-				return
-			}
-		}
-	}()
-}
-
-func (g *globalLock) refreshExec() bool {
-	script := `
-	local token = redis.call('get',KEYS[1])
-	if token == ARGV[1]
-	then
-		redis.call('set',KEYS[1],ARGV[1],'EX',ARGV[2])
-		return 'OK'
-	end
-		return 'ERROR'
-	`
-
-	resp, err := g.redis.Eval(g.ctx, script, []string{g.uniqueKey}, g.value, 5).Result()
-	if resp != "OK" {
-		log.Println("globalLock refresh", resp, err, g.uniqueKey, g.value)
-	}
-	return resp == "OK"
-}

lockx/lockx_test.go

@@ -1,52 +0,0 @@
-package lockx_test
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"code.yun.ink/open/timer/lockx"
-	"github.com/go-redis/redis/v8"
-)
-
-var Redis *redis.Client
-
-// func TestMain(m *testing.M) {
-// 	client := redis.NewClient(&redis.Options{
-// 		Addr:     "127.0.0.1" + ":" + "6379",
-// 		Password: "", // no password set
-// 		DB:       0,  // use default DB
-// 	})
-// 	if client == nil {
-// 		fmt.Println("redis init error")
-// 		return
-// 	}
-// 	// fmt.Println("ffff")
-// 	Redis = client
-// }
-
-func TestLockx(t *testing.T) {
-	client := redis.NewClient(&redis.Options{
-		Addr:     "127.0.0.1" + ":" + "6379",
-		Password: "", // no password set
-		DB:       0,  // use default DB
-	})
-	if client == nil {
-		fmt.Println("redis init error")
-		return
-	}
-	fmt.Println("begin")
-	ctx := context.Background()
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	lock := lockx.NewGlobalLock(ctx, client, "lockx:test")
-
-	if !lock.Lock() {
-		fmt.Println("lock error")
-	}
-	defer lock.Unlock()
-
-	fmt.Println("ssss")
-
-}

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

once.go

@@ -0,0 +1,184 @@
+package timerx
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"runtime/debug"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/go-redis/redis/v8"
+)
+
+// 功能描述
+// 1. 任务可以多节点发布
+// 2. 每个任务的执行在全局仅会执行一次
+// 3. 任务执行失败支持快捷重新加入队列
+
+// 单次的任务队列
+type worker struct {
+	ctx     context.Context
+	zsetKey string
+	listKey string
+	redis   *redis.Client
+	worker  Callback
+}
+
+type WorkerCode int
+
+const (
+	WorkerCodeSuccess WorkerCode = 0  // 处理完成(不需要重入)
+	WorkerCodeAgain   WorkerCode = -1 // 需要继续定时,默认原来的时间
+)
+
+// 需要考虑执行失败重新放入队列的情况
+type Callback interface {
+	// 任务执行
+	// uniqueKey: 任务唯一标识
+	// jobType: 任务类型，用于区分任务
+	// data: 任务数据
+	Worker(jobType string, uniqueKey string, data interface{}) (WorkerCode, time.Duration)
+}
+
+var wo *worker = nil
+var once sync.Once
+
+type extendData struct {
+	Delay time.Duration
+	Data  interface{}
+}
+
+// 初始化
+func InitOnce(ctx context.Context, re *redis.Client, jobGlobalName string, jobCallback Callback) *worker {
+
+	once.Do(func() {
+		wo = &worker{
+			ctx:     ctx,
+			zsetKey: "timer:once_zsetkey" + jobGlobalName,
+			listKey: "timer:once_listkey" + jobGlobalName,
+			redis:   re,
+			worker:  jobCallback,
+		}
+		go wo.getTask()
+		go wo.watch()
+	})
+
+	return wo
+}
+
+// 添加任务
+// 重复插入就代表覆盖
+func (w *worker) Add(jobType string, uniqueKey string, delayTime time.Duration, data interface{}) error {
+	if delayTime.Abs() != delayTime {
+		return fmt.Errorf("时间间隔不能为负数")
+	}
+	if delayTime == 0 {
+		return fmt.Errorf("时间间隔不能为0")
+	}
+
+	redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
+
+	ed := extendData{
+		Delay: delayTime,
+		Data:  data,
+	}
+	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,
+	}).Result()
+
+	return err
+}
+
+// 删除任务
+func (w *worker) Del(jobType string, uniqueKey string) error {
+	redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
+
+	w.redis.Del(w.ctx, redisKey).Result()
+
+	w.redis.ZRem(w.ctx, w.zsetKey, redisKey).Result()
+
+	return nil
+}
+
+// 获取任务
+func (w *worker) getTask() {
+	timer := time.NewTicker(time.Millisecond * 200)
+	defer timer.Stop()
+
+Loop:
+	for {
+		select {
+		case <-timer.C:
+			script := `
+				local token = redis.call('zrangebyscore',KEYS[1],ARGV[1],ARGV[2])
+				for i,v in ipairs(token) do
+					redis.call('zrem',KEYS[1],v)
+					redis.call('lpush',KEYS[2],v)
+				end
+				return "OK"
+				`
+			w.redis.Eval(w.ctx, script, []string{w.zsetKey, w.listKey}, 0, time.Now().UnixMilli()).Result()
+			// fmt.Println(i, err)
+
+		case <-w.ctx.Done():
+			break Loop
+		}
+	}
+}
+
+// 监听任务
+func (w *worker) watch() {
+	for {
+		keys, err := w.redis.BLPop(w.ctx, time.Second*10, w.listKey).Result()
+		if err != nil {
+			fmt.Println("watch err:", err)
+			continue
+		}
+
+		go w.doTask(keys[1])
+
+	}
+}
+
+func (w *worker) doTask(key string) {
+	defer func() {
+		if err := recover(); err != nil {
+			fmt.Println("timer:定时器出错", err)
+			log.Println("errStack", string(debug.Stack()))
+		}
+	}()
+
+	s := strings.Split(key, "[:]")
+
+	// 读取数据
+	str, err := w.redis.Get(w.ctx, key).Result()
+	if err != nil {
+		fmt.Println("execJob err:", 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)
+
+	if code == WorkerCodeAgain {
+		// 重新放入队列
+		fmt.Println("重入时间：", time.Now().Format("2006-01-02 15:04:05"))
+		if t != 0 && t == t.Abs() {
+			ed.Delay = t
+		}
+		w.Add(s[0], s[1], ed.Delay, ed.Data)
+	}
+}

once_test.go

@@ -0,0 +1,2 @@
+package timerx
+

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

@@ -1,4 +1,4 @@
-package timer
+package timerx
 
 // 作者：黄新云
 
@@ -13,30 +13,23 @@ import (
 )
 
 // 定时器
-// 原理：每毫秒的时间触发
+// 1. 这个定时器的作用范围是本机
 
 // uuid -> timerStr
 var timerMap = make(map[string]*timerStr)
 var timerMapMux sync.Mutex
 
-var timerCount int        // 当前定时数目
-var onceLimit sync.Once   // 实现单例
-var nextTime = time.Now() // 下一次执行的时间
+var timerCount int      // 当前定时数目
+var onceLimit sync.Once // 实现单例
 
-type ContextValueKey string // 定义context 传递的Key类型
+type Single struct{}
 
-const (
-	extendParamKey ContextValueKey = "extend_param"
-)
-
-type single struct{}
-
-var sin *single = nil
+var sin *Single = nil
 
 // 定时器类
-func InitSingle(ctx context.Context) *single {
+func InitSingle(ctx context.Context) *Single {
 	onceLimit.Do(func() {
-		sin = &single{}
+		sin = &Single{}
 
 		timer := time.NewTicker(time.Millisecond * 200)
 		go func(ctx context.Context) {
@@ -49,7 +42,7 @@ func InitSingle(ctx context.Context) *single {
 						continue
 					}
 					// 迭代定时器
-					sin.iteratorTimer(ctx, t)
+					sin.iterator(ctx, t)
 					// fmt.Println("timer: 执行")
 				case <-ctx.Done():
 					// 跳出循环
@@ -64,7 +57,12 @@ func InitSingle(ctx context.Context) *single {
 }
 
 // 间隔定时器
-func (s *single) AddTimer(space time.Duration, call callback, extend ExtendParams) (int, error) {
+// @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()
 
@@ -82,8 +80,7 @@ func (s *single) AddTimer(space time.Duration, call callback, extend ExtendParam
 		NextTime:   nowTime, // nowTime.Add(space), // 添加任务的时候就执行一次
 		SpaceTime:  space,
 		CanRunning: make(chan struct{}, 1),
-		UniqueKey:  "",
-		Extend:     extend,
+		ExtendData: extend,
 	}
 
 	timerMap[fmt.Sprintf("%d", timerCount)] = &t
@@ -96,21 +93,15 @@ func (s *single) AddTimer(space time.Duration, call callback, extend ExtendParam
 	return timerCount, nil
 }
 
-// 添加需要定时的规则
-func (s *single) AddToTimer(space time.Duration, call callback) int {
-	extend := ExtendParams{}
-	count, _ := s.AddTimer(space, call, extend)
-	return count
-}
-
-func (s *single) DelToTimer(index string) {
+// 删除定时器
+func (s *Single) Del(index string) {
 	timerMapMux.Lock()
 	defer timerMapMux.Unlock()
 	delete(timerMap, index)
 }
 
 // 迭代定时器列表
-func (s *single) iteratorTimer(ctx context.Context, nowTime time.Time) {
+func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
 	timerMapMux.Lock()
 	defer timerMapMux.Unlock()
 
@@ -159,7 +150,7 @@ func (s *single) iteratorTimer(ctx context.Context, nowTime time.Time) {
 						}
 					}()
 					// fmt.Printf("timer: 准备执行 %v %v \n", k, v.Tag)
-					s.timerAction(ctx, v.Callback, v.UniqueKey, v.Extend)
+					s.doTask(ctx, v.Callback, v.ExtendData)
 				default:
 					// fmt.Printf("timer: 已在执行 %v %v \n", k, v.Tag)
 					return
@@ -182,33 +173,14 @@ func (s *single) iteratorTimer(ctx context.Context, nowTime time.Time) {
 	// fmt.Println("timer: one finish")
 }
 
-// 定义各个回调函数
-type callback func(ctx context.Context) bool
-
 // 定时器操作类
 // 这里不应painc
-func (s *single) timerAction(ctx context.Context, call callback, uniqueKey string, extend ExtendParams) bool {
+func (s *Single) doTask(ctx context.Context, call callback, extend interface{}) error {
 	defer func() {
 		if err := recover(); err != nil {
 			fmt.Println("timer:定时器出错", err)
 			log.Println("errStack", string(debug.Stack()))
 		}
 	}()
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
-
-	// 附加数据
-	ctx = context.WithValue(ctx, extendParamKey, extend)
-
-	return call(ctx)
-}
-
-// 快捷方法
-func GetExtendParams(ctx context.Context) (*ExtendParams, error) {
-	val := ctx.Value(extendParamKey)
-	params, ok := val.(ExtendParams)
-	if !ok {
-		return nil, errors.New("没找到参数")
-	}
-	return &params, nil
+	return call(ctx, extend)
 }

single_test.go

@@ -1,15 +1,20 @@
-package timer_test
+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

@@ -1,6 +1,9 @@
-package timer
+package timerx
 
-import "time"
+import (
+	"context"
+	"time"
+)
 
 type timerStr struct {
 	Callback   callback        // 需要回调的方法
@@ -8,11 +11,34 @@ type timerStr struct {
 	BeginTime  time.Time       // 初始化任务的时间
 	NextTime   time.Time       // [删]下一次执行的时间
 	SpaceTime  time.Duration   // 任务间隔时间
-	UniqueKey  string          // 全局唯一键
-	Extend     ExtendParams    // 附加参数
+	TaskId     string          // 任务ID 全局唯一键
+	ExtendData interface{}     // 附加参数
+	JobType    JobType         // 任务类型
+	JobData    *JobData        // 任务时间数据
 }
 
-// 扩展参数
-type ExtendParams struct {
-	Params map[string]interface{} // 带出去的参数
+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

worker.go

@@ -1,159 +0,0 @@
-package timer
-
-import (
-	"context"
-	"encoding/json"
-	"fmt"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/go-redis/redis/v8"
-)
-
-// 单次的任务队列
-type worker struct {
-	ctx     context.Context
-	zsetKey string
-	listKey string
-	redis   *redis.Client
-	worker  WorkerInterface
-}
-
-type WorkerCode int
-
-const (
-	WorkerCodeSuccess WorkerCode = 0
-	WorkerCodeAgain   WorkerCode = -1
-)
-
-// 需要考虑执行失败重新放入队列的情况
-type WorkerInterface interface {
-	Worker(uniqueKey string, jobType string, data map[string]interface{}) WorkerCode
-}
-
-var wo *worker = nil
-var once sync.Once
-
-type extendData struct {
-	Delay time.Duration
-	Data  map[string]interface{}
-}
-
-func InitWorker(ctx context.Context, re *redis.Client, w WorkerInterface) *worker {
-
-	once.Do(func() {
-		wo = &worker{
-			ctx:     ctx,
-			zsetKey: "timer:job_zsetkey",
-			listKey: "timer:job_listkey",
-			redis:   re,
-			worker:  w,
-		}
-		go wo.getTask()
-		go wo.execTask()
-	})
-
-	return wo
-}
-
-// 添加任务
-// 重复插入就代表覆盖
-func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration, data map[string]interface{}) error {
-	if delayTime.Abs() != delayTime {
-		return fmt.Errorf("时间间隔不能为负数")
-	}
-	if delayTime == 0 {
-		return fmt.Errorf("时间间隔不能为0")
-	}
-
-	redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType)
-
-	ed := extendData{
-		Delay: delayTime,
-		Data:  data,
-	}
-	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,
-	}).Result()
-
-	return err
-}
-
-// 删除任务
-func (w *worker) Del(uniqueKey string, jobType string) error {
-	redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType)
-
-	w.redis.Del(w.ctx, redisKey).Result()
-
-	w.redis.ZRem(w.ctx, w.zsetKey, redisKey).Result()
-
-	return nil
-}
-
-// 获取任务
-func (w *worker) getTask() {
-	timer := time.NewTicker(time.Millisecond * 200)
-	defer timer.Stop()
-
-Loop:
-	for {
-		select {
-		case <-timer.C:
-			script := `
-				local token = redis.call('zrangebyscore',KEYS[1],ARGV[1],ARGV[2])
-				for i,v in ipairs(token) do
-					redis.call('zrem',KEYS[1],v)
-					redis.call('lpush',KEYS[2],v)
-				end
-				return "OK"
-				`
-			w.redis.Eval(w.ctx, script, []string{w.zsetKey, w.listKey}, 0, time.Now().UnixMilli()).Result()
-			// fmt.Println(i, err)
-
-		case <-w.ctx.Done():
-			break Loop
-		}
-	}
-}
-
-// 执行任务
-func (w *worker) execTask() {
-	for {
-		keys, err := w.redis.BLPop(w.ctx, time.Second*10, w.listKey).Result()
-		if err != nil {
-			fmt.Println("watch err:", err)
-			continue
-		}
-
-		go func() {
-			s := strings.Split(keys[1], "[:]")
-
-			// 读取数据
-			str, err := w.redis.Get(w.ctx, keys[1]).Result()
-			if err != nil {
-				fmt.Println("execJob err:", err)
-				return
-			}
-			ed := extendData{}
-			json.Unmarshal([]byte(str), &ed)
-
-			fmt.Println("开始时间：", time.Now().Format("2006-01-02 15:04:05"))
-			code := w.worker.Worker(s[0], s[1], ed.Data)
-
-			if code == WorkerCodeAgain {
-				// 重新放入队列
-				fmt.Println("重入时间：", time.Now().Format("2006-01-02 15:04:05"))
-				w.Add(s[0], s[1], ed.Delay, ed.Data)
-			}
-		}()
-	}
-}