修改go.mod为github

cluster.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"fmt"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -20,8 +19,6 @@ import (
 
 // 暂不支持删除定时器，因为这个定时器的设计是基于全局的，如果删除了，那么其他服务就不知道了
 
-// TODO:如果获取到任务不能处理的，应放回队列（因为可能新旧代码同时上线，新的添加了任务处理）
-
 // 单例模式
 var clusterOnceLimit sync.Once
 
@@ -208,7 +205,6 @@ func (c *Cluster) addJob(ctx context.Context, taskId string, beginTime time.Time
 // TODO:注册的任务需放在Redis集中存储，因为本地的话，如果有多个服务，那么就会出现不一致的情况。但是要注意服务如何进行下线，由于是主动上报的，需要有一个机制进行删除过期的任务（添加任务&定时器轮训注册）
 func (c *Cluster) getNextTime() {
 
-	// log.Println("begin computer")
 	ctx, cancel := context.WithCancel(c.ctx)
 	defer cancel()
 
@@ -294,12 +290,14 @@ 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)
+				if err != redis.Nil {
+					c.logger.Errorf(c.ctx, "BLPop watch err:%+v", err)
+				}
 				continue
 			}
 			_, ok := clusterWorkerList.Load(keys[1])
 			if !ok {
-				fmt.Println("watch timer:任务不存在", keys[1])
+				c.logger.Errorf(c.ctx, "watch timer:任务不存在", keys[1])
 				c.redis.SAdd(c.ctx, c.setKey, keys[1])
 				continue
 			}
@@ -310,18 +308,18 @@ func (c *Cluster) watch() {
 	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)
+				} else {
+					c.logger.Errorf(c.ctx, "SPop watch err:%+v", err)
 				}
 				continue
 			}
 			_, ok := clusterWorkerList.Load(taskId)
 			if !ok {
-				fmt.Println("watch timer:任务不存在", taskId)
+				c.logger.Errorf(c.ctx, "watch timer:任务不存在", taskId)
 				c.redis.SAdd(c.ctx, c.setKey, taskId)
 				continue
 			}

go.mod

@@ -1,4 +1,4 @@
-module code.yun.ink/pkg/timerx
+module github.com/yuninks/timerx
 
 go 1.19
 

option.go

@@ -20,8 +20,16 @@ func newOptions(opts ...Option) Options {
 	return o
 }
 
+// 设置日志
 func SetLogger(log Logger) Option {
 	return func(o *Options) {
 		o.logger = log
 	}
 }
+
+// 设定时区
+func SetTimeZone(zone string) Option {
+	return func(o *Options) {
+		// todo
+	}
+}

single.go

@@ -6,7 +6,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"sync"
 	"time"
@@ -22,14 +21,23 @@ var timerMapMux sync.Mutex
 var timerCount int      // 当前定时数目
 var onceLimit sync.Once // 实现单例
 
-type Single struct{}
+type Single struct {
+	ctx    context.Context
+	logger Logger
+}
 
 var sin *Single = nil
 
 // 定时器类
-func InitSingle(ctx context.Context) *Single {
+func InitSingle(ctx context.Context, opts ...Option) *Single {
 	onceLimit.Do(func() {
-		sin = &Single{}
+
+		op := newOptions(opts...)
+
+		sin = &Single{
+			ctx:    ctx,
+			logger: op.logger,
+		}
 
 		timer := time.NewTicker(time.Millisecond * 200)
 		go func(ctx context.Context) {
@@ -49,7 +57,7 @@ func InitSingle(ctx context.Context) *Single {
 					break Loop
 				}
 			}
-			log.Println("timer: initend")
+			sin.logger.Infof(ctx, "timer: initend")
 		}(ctx)
 	})
 
@@ -67,6 +75,7 @@ func (s *Single) Add(space time.Duration, call callback, extend interface{}) (in
 	defer timerMapMux.Unlock()
 
 	if space != space.Abs() {
+		s.logger.Errorf(s.ctx, "space must be positive")
 		return 0, errors.New("space must be positive")
 	}
 
@@ -178,8 +187,7 @@ func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
 func (s *Single) doTask(ctx context.Context, call callback, extend interface{}) error {
 	defer func() {
 		if err := recover(); err != nil {
-			fmt.Println("timer:定时器出错", err)
+			s.logger.Errorf(ctx, "timer:定时器出错 err:%+v stack:%s", err, string(debug.Stack()))
-			log.Println("errStack", string(debug.Stack()))
 		}
 	}()
 	return call(ctx, extend)