去掉单例模式，改为每次初始化均生成一个新的实例

2025-08-23 18:55:59 +08:00
parent 2a7092ab0a
commit bb5b01071a
4 changed files with 110 additions and 106 deletions
@@ -25,7 +25,7 @@ import (
 // 暂不支持删除定时器，因为这个定时器的设计是基于全局的，如果删除了，那么其他服务就不知道了

 // 单例模式
-var clusterOnceLimit sync.Once
+// var clusterOnceLimit sync.Once

 // 已注册的任务列表
 var clusterWorkerList sync.Map
@@ -48,69 +48,69 @@ type Cluster struct {
 	priorityKey string // 全局优先级的key
 }

-var clu *Cluster = nil
+// var clu *Cluster = nil

 // 初始化定时器
 // 全局只需要初始化一次
 func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix string, opts ...Option) *Cluster {

-	clusterOnceLimit.Do(func() {
-		op := newOptions(opts...)
+	// clusterOnceLimit.Do(func() {
+	op := newOptions(opts...)

-		clu = &Cluster{
-			ctx:         ctx,
-			redis:       red,
-			cache:       cachex.NewCache(),
-			timeout:     op.timeout,
-			logger:      op.logger,
-			keyPrefix:   keyPrefix,
-			location:    op.location,
-			priority:    op.priority,
-			lockKey:     "timer:cluster_globalLockKey" + keyPrefix, // 定时器的全局锁
-			zsetKey:     "timer:cluster_zsetKey" + keyPrefix,       // 有序集合
-			listKey:     "timer:cluster_listKey" + keyPrefix,       // 列表
-			setKey:      "timer:cluster_setKey" + keyPrefix,        // 重入集合
-			priorityKey: "timer:cluster_priorityKey" + keyPrefix,   // 全局优先级的key
+	clu := &Cluster{
+		ctx:         ctx,
+		redis:       red,
+		cache:       cachex.NewCache(),
+		timeout:     op.timeout,
+		logger:      op.logger,
+		keyPrefix:   keyPrefix,
+		location:    op.location,
+		priority:    op.priority,
+		lockKey:     "timer:cluster_globalLockKey" + keyPrefix, // 定时器的全局锁
+		zsetKey:     "timer:cluster_zsetKey" + keyPrefix,       // 有序集合
+		listKey:     "timer:cluster_listKey" + keyPrefix,       // 列表
+		setKey:      "timer:cluster_setKey" + keyPrefix,        // 重入集合
+		priorityKey: "timer:cluster_priorityKey" + keyPrefix,   // 全局优先级的key
+	}
+
+	// 设置锁的超时时间
+	lockx.InitOption(lockx.SetTimeout(op.timeout))
+
+	// 监听任务
+	go clu.watch()
+
+	priorityTime := time.NewTicker(time.Second * 10)
+	go func(ctx context.Context) {
+		clu.setPriority()
+	Loop:
+		for {
+			select {
+			case <-priorityTime.C:
+				clu.setPriority()
+			case <-ctx.Done():
+				break Loop
+			}
 		}
+	}(ctx)

-		// 设置锁的超时时间
-		lockx.InitOption(lockx.SetTimeout(op.timeout))
+	timer := time.NewTicker(time.Millisecond * 200)

-		// 监听任务
-		go clu.watch()
-
-		priorityTime := time.NewTicker(time.Second * 10)
-		go func(ctx context.Context) {
-			clu.setPriority()
-		Loop:
-			for {
-				select {
-				case <-priorityTime.C:
-					clu.setPriority()
-				case <-ctx.Done():
-					break Loop
+	go func(ctx context.Context) {
+	Loop:
+		for {
+			select {
+			case <-timer.C:
+				if !clu.canRun() {
+					continue
 				}
+				clu.getTask()
+				clu.getNextTime()
+			case <-ctx.Done():
+				break Loop
 			}
-		}(ctx)
-
-		timer := time.NewTicker(time.Millisecond * 200)
-
-		go func(ctx context.Context) {
-		Loop:
-			for {
-				select {
-				case <-timer.C:
-					if !clu.canRun() {
-						continue
-					}
-					clu.getTask()
-					clu.getNextTime()
-				case <-ctx.Done():
-					break Loop
-				}
-			}
-		}(ctx)
-	})
+		}
+	}(ctx)
+	// })
 	return clu
 }

@@ -181,7 +181,7 @@ func (l *Cluster) setPriority() bool {
 	`
 	priority := fmt.Sprintf("%d", l.priority)

-	expireTime := (time.Second*30).Seconds() // 设置过期时间为1分钟
+	expireTime := (time.Second * 30).Seconds() // 设置过期时间为1分钟
 	res, err := l.redis.Eval(l.ctx, script, []string{l.priorityKey}, priority, expireTime).Result()
 	if err != nil {
 		l.logger.Errorf(l.ctx, "设置全局优先级失败:%s", err.Error())
@@ -24,8 +24,8 @@ func main() {

 	// re()
 	// d()
-	cluster()
-	// once()
+	// cluster()
+	once()

 	select {}

@@ -46,11 +46,12 @@ func once() {
 	if err != nil {
 		fmt.Println(err)
 	}
-	d = OnceData{
-		Num: 4,
-	}
+	// d = OnceData{
+	// 	Num: 4,
+	// }
+	dd := 123
 	// dy, _ = json.Marshal(d)
-	err = one.Create("test", "test4", 1*time.Second, d)
+	err = one.Save("test", "test4", 1*time.Second, dd)
 	if err != nil {
 		fmt.Println(err)
 	}
@@ -67,7 +68,11 @@ func (l OnceWorker) Worker(ctx context.Context, taskType string, taskId string,
 	fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
 	fmt.Println(taskType, taskId)

-	fmt.Printf("原来的参数：%+v\n", attachData)
+	fmt.Printf("原来的参数：%+v %T\n", attachData, attachData)
+
+	v, ok := attachData.(int64)
+	fmt.Println("vvvvvvv", v, ok)
+	// fmt.Printf()

 	// d := OnceData{}

@@ -89,7 +94,7 @@ func (l OnceWorker) Worker(ctx context.Context, taskType string, taskId string,
 func cluster() {
 	client := getRedis()
 	ctx := context.Background()
-	cluster := timerx.InitCluster(ctx, client, "test",timerx.SetPriority(101))
+	cluster := timerx.InitCluster(ctx, client, "test", timerx.SetPriority(101))
 	err := cluster.EverySpace(ctx, "test_space", 1*time.Second, aa, "这是秒任务")
 	fmt.Println(err)
 	err = cluster.EveryMinute(ctx, "test_min", 15, aa, "这是分钟任务")
@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"runtime/debug"
 	"strings"
-	"sync"
 	"time"

 	"github.com/go-redis/redis/v8"
@@ -47,8 +46,8 @@ type Callback interface {
 	Worker(ctx context.Context, taskType string, taskId string, attachData interface{}) *OnceWorkerResp
 }

-var wo *Once = nil
-var once sync.Once
+// var wo *Once = nil
+// var once sync.Once

 type extendData struct {
 	Delay time.Duration
@@ -62,19 +61,19 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 	}

 	op := newOptions(opts...)
-	once.Do(func() {
-		wo = &Once{
-			ctx:       ctx,
-			logger:    op.logger,
-			zsetKey:   "timer:once_zsetkey" + keyPrefix,
-			listKey:   "timer:once_listkey" + keyPrefix,
-			redis:     re,
-			worker:    call,
-			keyPrefix: keyPrefix,
-		}
-		go wo.getTask()
-		go wo.watch()
-	})
+	// once.Do(func() {
+	wo := &Once{
+		ctx:       ctx,
+		logger:    op.logger,
+		zsetKey:   "timer:once_zsetkey" + keyPrefix,
+		listKey:   "timer:once_listkey" + keyPrefix,
+		redis:     re,
+		worker:    call,
+		keyPrefix: keyPrefix,
+	}
+	go wo.getTask()
+	go wo.watch()
+	// })

 	return wo
 }
@@ -22,7 +22,7 @@ var singleWorkerList sync.Map

 var singleTimerIndex int // 当前定时数目

-var singleOnceLimit sync.Once // 实现单例
+// var singleOnceLimit sync.Once // 实现单例

 type Single struct {
 	ctx      context.Context
@@ -30,7 +30,7 @@ type Single struct {
 	location *time.Location
 }

-var sin *Single = nil
+// var sin *Single = nil

 var singleNextTime = time.Now() // 下一次执行的时间

@@ -38,36 +38,36 @@ var singleNextTime = time.Now() // 下一次执行的时间
 // @param ctx context.Context 上下文
 // @param opts ...Option 配置项
 func InitSingle(ctx context.Context, opts ...Option) *Single {
-	singleOnceLimit.Do(func() {
-		op := newOptions(opts...)
+	// singleOnceLimit.Do(func() {
+	op := newOptions(opts...)

-		sin = &Single{
-			ctx:      ctx,
-			logger:   op.logger,
-			location: op.location,
-		}
+	sin := &Single{
+		ctx:      ctx,
+		logger:   op.logger,
+		location: op.location,
+	}

-		timer := time.NewTicker(time.Millisecond * 200)
-		go func(ctx context.Context) {
-		Loop:
-			for {
-				select {
-				case t := <-timer.C:
-					if t.Before(singleNextTime) {
-						// 当前时间小于下次发送时间：跳过
-						continue
-					}
-					// 迭代定时器
-					sin.iterator(ctx)
-					// fmt.Println("timer: 执行")
-				case <-ctx.Done():
-					// 跳出循环
-					break Loop
+	timer := time.NewTicker(time.Millisecond * 200)
+	go func(ctx context.Context) {
+	Loop:
+		for {
+			select {
+			case t := <-timer.C:
+				if t.Before(singleNextTime) {
+					// 当前时间小于下次发送时间：跳过
+					continue
 				}
+				// 迭代定时器
+				sin.iterator(ctx)
+				// fmt.Println("timer: 执行")
+			case <-ctx.Done():
+				// 跳出循环
+				break Loop
 			}
-			sin.logger.Infof(ctx, "timer: initend")
-		}(ctx)
-	})
+		}
+		sin.logger.Infof(ctx, "timer: initend")
+	}(ctx)
+	// })

 	return sin
 }