添加心跳相关内容

cluster.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"runtime/debug"
+	"strconv"
 	"sync"
 	"time"
 
@@ -31,21 +32,24 @@ type Cluster struct {
 	keyPrefix string                // key前缀
 	location  *time.Location        // 根据时区计算的时间
 
-	lockKey string // 全局计算的key
+	lockKey      string // 全局计算的key
-	zsetKey string // 有序集合的key
+	zsetKey      string // 有序集合的key
-	listKey string // 可执行的任务列表的key
+	listKey      string // 可执行的任务列表的key
-	setKey  string // 重入集合的key
+	setKey       string // 重入集合的key
+	heartbeatKey string // 心跳的Key
+	leaderKey    string // 上报当前的Leader
 
 	priority    *priority.Priority // 全局优先级
 	priorityKey string             // 全局优先级的key
 	usePriority bool               // 是否使用优先级
 
-	wg         sync.WaitGroup // 等待组
+	wg               sync.WaitGroup // 等待组
-	isLeader   bool           // 是否是领导
+	isLeader         bool           // 是否是领导
-	leaderLock sync.RWMutex   // 领导锁
+	leaderLock       sync.RWMutex   // 领导锁
-	leaderKey  string         // 实例Id
+	leaderUniLockKey string         // 领导唯一锁
-	workerList sync.Map       // 注册的任务列表
+	workerList       sync.Map       // 注册的任务列表
-	stopChan   chan struct{}
+	stopChan         chan struct{}  //
+	instanceId       string         // 实例ID
 }
 
 // 初始化定时器
@@ -57,23 +61,28 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 
 	ctx, cancel := context.WithCancel(ctx)
 
+	U, _ := uuid.NewV7()
+
 	clu := &Cluster{
-		ctx:         ctx,
+		ctx:              ctx,
-		cancel:      cancel,
+		cancel:           cancel,
-		redis:       red,
+		redis:            red,
-		cache:       cachex.NewCache(),
+		cache:            cachex.NewCache(),
-		timeout:     op.timeout,
+		timeout:          op.timeout,
-		logger:      op.logger,
+		logger:           op.logger,
-		keyPrefix:   keyPrefix,
+		keyPrefix:        keyPrefix,
-		location:    op.location,
+		location:         op.location,
-		lockKey:     "timer:cluster_globalLockKey" + keyPrefix, // 定时器的全局锁
+		lockKey:          "timer:cluster_globalLockKey" + keyPrefix,    // 定时器的全局锁
-		zsetKey:     "timer:cluster_zsetKey" + keyPrefix,       // 有序集合
+		zsetKey:          "timer:cluster_zsetKey" + keyPrefix,          // 有序集合
-		listKey:     "timer:cluster_listKey" + keyPrefix,       // 列表
+		listKey:          "timer:cluster_listKey" + keyPrefix,          // 列表
-		setKey:      "timer:cluster_setKey" + keyPrefix,        // 重入集合
+		setKey:           "timer:cluster_setKey" + keyPrefix,           // 重入集合
-		priorityKey: "timer:cluster_priorityKey" + keyPrefix,   // 全局优先级的key
+		priorityKey:      "timer:cluster_priorityKey" + keyPrefix,      // 全局优先级的key
-		leaderKey:   "timer:cluster_leaderKey" + keyPrefix,     // 领导
+		leaderUniLockKey: "timer:cluster_leaderUniLockKey" + keyPrefix, // 领导唯一锁
-		usePriority: op.usePriority,
+		leaderKey:        "timer:cluster_leaderKey" + keyPrefix,        // 上报当前Leader
-		stopChan:    make(chan struct{}),
+		heartbeatKey:     "timer:cluster_heartbeatKey" + keyPrefix,     // 心跳 有序集合
+		usePriority:      op.usePriority,
+		stopChan:         make(chan struct{}),
+		instanceId:       U.String(),
 	}
 
 	// 初始化优先级
@@ -85,13 +94,16 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 	// 启动守护进程
 	clu.startDaemon()
 
+	clu.logger.Infof(ctx, "InitCluster success keyPrefix:%s instanceId:%s", clu.keyPrefix, clu.instanceId)
+
 	return clu
 }
 
 // Stop 停止集群定时器
-func (c *Cluster) Stop() {
+func (l *Cluster) Stop() {
-	close(c.stopChan)
+	close(l.stopChan)
-	c.wg.Wait()
+	l.cleanHeartbeat(true)
+	l.wg.Wait()
 }
 
 // 守护任务
@@ -101,6 +113,14 @@ func (l *Cluster) startDaemon() {
 	l.wg.Add(1)
 	go l.leaderElection()
 
+	// 心跳上报
+	l.wg.Add(1)
+	go l.heartbeatLoop()
+
+	// 心跳过期清理
+	l.wg.Add(1)
+	go l.cleanHeartbeatLoop()
+
 	// 任务调度
 	l.wg.Add(1)
 	go l.scheduleTasks()
@@ -111,6 +131,76 @@ func (l *Cluster) startDaemon() {
 
 }
 
+// 心跳上报
+// 需要确定当前存活的实例&当前实例是否是领导
+func (l *Cluster) heartbeatLoop() {
+	defer l.wg.Done()
+
+	// 先执行一次
+	l.heartbeat()
+
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-l.stopChan:
+			return
+		case <-l.ctx.Done():
+			return
+		case <-ticker.C:
+			l.heartbeat()
+		}
+	}
+
+}
+
+// 单次心跳
+func (l *Cluster) heartbeat() error {
+	err := l.redis.ZAdd(l.ctx, l.heartbeatKey, &redis.Z{
+		Score:  float64(time.Now().UnixMilli()),
+		Member: l.instanceId,
+	}).Err()
+
+	if err != nil {
+		l.logger.Errorf(l.ctx, "heartbeat redis.ZAdd err:%v", err)
+		return err
+	}
+
+	return nil
+}
+
+// 心跳清理（leader可操作）
+func (l *Cluster) cleanHeartbeatLoop() {
+	defer l.wg.Done()
+
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-l.stopChan:
+			return
+		case <-l.ctx.Done():
+			return
+		case <-ticker.C:
+			if l.isCurrentLeader() {
+				l.cleanHeartbeat(false)
+			}
+		}
+	}
+
+}
+
+// 单次清理
+func (l *Cluster) cleanHeartbeat(cleanSelf bool) error {
+
+	if cleanSelf {
+		return l.redis.ZRem(l.ctx, l.heartbeatKey, l.instanceId).Err()
+	}
+	return l.redis.ZRemRangeByScore(l.ctx, l.heartbeatKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Second).UnixMilli(), 10)).Err()
+}
+
 // 领导选举
 // 领导作用：全局推选一个人计算执行时间&移入队列，避免每个都进行计算浪费资源
 func (l *Cluster) leaderElection() {
@@ -147,7 +237,7 @@ func (l *Cluster) getLeaderLock() error {
 	defer cancel()
 
 	// 尝试加锁
-	lock, err := lockx.NewGlobalLock(ctx, l.redis, l.leaderKey)
+	lock, err := lockx.NewGlobalLock(ctx, l.redis, l.leaderUniLockKey)
 	if err != nil {
 		l.logger.Errorf(l.ctx, "getLeaderLock err:%+v", err)
 		return err
@@ -166,6 +256,9 @@ func (l *Cluster) getLeaderLock() error {
 	l.isLeader = true
 	l.leaderLock.Unlock()
 
+	// 上报当前的Leader实例
+	l.redis.Set(l.ctx, l.leaderKey, l.instanceId, time.Hour*24)
+
 	l.logger.Infof(l.ctx, "getLeaderLock Instance %s became leader", lock.GetValue())
 
 	go func() {
@@ -417,7 +510,7 @@ func (l *Cluster) calculateNextTimes() {
 			return 1
 			`
 
-		lockKey := fmt.Sprintf("%s:lock:calc:%s:%d", l.keyPrefix, val.TaskId, nextTime.UnixMilli())
+		lockKey := fmt.Sprintf("%s_%s_%d", l.keyPrefix, val.TaskId, nextTime.UnixMilli())
 		_, err = pipe.Eval(l.ctx, script, []string{l.zsetKey, lockKey},
 			nextTime.UnixMilli(), val.TaskId, 60).Result()
 		if err != nil {

go.mod

@@ -1,6 +1,6 @@
 module github.com/yuninks/timerx
 
-go 1.21
+go 1.24
 
 require (
 	github.com/go-redis/redis/v8 v8.11.5
@@ -8,7 +8,7 @@ require (
 	github.com/satori/go.uuid v1.2.0
 	github.com/stretchr/testify v1.11.1
 	github.com/yuninks/cachex v1.0.5
-	github.com/yuninks/lockx v1.1.0
+	github.com/yuninks/lockx v1.1.1
 )
 
 require (

go.sum

@@ -31,8 +31,8 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/yuninks/cachex v1.0.5 h1:Y2NmTsuEgwEVYb7FVFh5tUN67kmrUioeksQqLbOAwsM=
 github.com/yuninks/cachex v1.0.5/go.mod h1:5357qz18UvHTJSgZzkMamUzZoFzGeKG9+4tIUBXRSVM=
-github.com/yuninks/lockx v1.1.0 h1:Qkf+RqYQ6Vr242h3dWxWJqIvIp9/6j5ijTkbuWvrgqU=
+github.com/yuninks/lockx v1.1.1 h1:b687xJrBblvquP9czP3iiW62VPr/RerJbQMIi0zQizs=
-github.com/yuninks/lockx v1.1.0/go.mod h1:Y/MtD+4Zc79/0Qsdd9c99jxW5WWxXVZlxcJyYJFJy5A=
+github.com/yuninks/lockx v1.1.1/go.mod h1:tM46x/fp9046YnTIeddmUhkxFjJ/f0g4J+1zdaGKfd8=
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0=
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
 golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=