pkg/timerx
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化全局单次定时器的冲突问题

Browse Source
This commit is contained in:
Yun
2023-12-27 17:11:33 +08:00
parent 362d1f455a
commit 1beafa934c
1 changed files with 9 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
once.go
+9 -9
View File
@@ -40,7 +40,7 @@ type Callback interface {
// uniqueKey: 任务唯一标识 // uniqueKey: 任务唯一标识
// jobType: 任务类型，用于区分任务 // jobType: 任务类型，用于区分任务
// data: 任务数据 // data: 任务数据
Worker(uniqueKey string, jobType string, data interface{}) (WorkerCode, time.Duration) Worker(jobType string, uniqueKey string, data interface{}) (WorkerCode, time.Duration)
} }
var wo *worker = nil var wo *worker = nil
@@ -52,15 +52,15 @@ type extendData struct {
} }
// 初始化 // 初始化
func InitOnce(ctx context.Context, re *redis.Client, w Callback) *worker { func InitOnce(ctx context.Context, re *redis.Client, jobGlobalName string, jobCallback Callback) *worker {
once.Do(func() { once.Do(func() {
wo = &worker{ wo = &worker{
ctx: ctx, ctx: ctx,
zsetKey: "timer:once_zsetkey", zsetKey: "timer:once_zsetkey" + jobGlobalName,
listKey: "timer:once_listkey", listKey: "timer:once_listkey" + jobGlobalName,
redis: re, redis: re,
worker: w, worker: jobCallback,
} }
go wo.getTask() go wo.getTask()
go wo.watch() go wo.watch()
@@ -71,7 +71,7 @@ func InitOnce(ctx context.Context, re *redis.Client, w Callback) *worker {
// 添加任务 // 添加任务
// 重复插入就代表覆盖 // 重复插入就代表覆盖
func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration, data interface{}) error { func (w *worker) Add(jobType string, uniqueKey string, delayTime time.Duration, data interface{}) error {
if delayTime.Abs() != delayTime { if delayTime.Abs() != delayTime {
return fmt.Errorf("时间间隔不能为负数") return fmt.Errorf("时间间隔不能为负数")
} }
@@ -79,7 +79,7 @@ func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration,
return fmt.Errorf("时间间隔不能为0") return fmt.Errorf("时间间隔不能为0")
} }
redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType) redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
ed := extendData{ ed := extendData{
Delay: delayTime, Delay: delayTime,
@@ -101,8 +101,8 @@ func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration,
} }
// 删除任务 // 删除任务
func (w *worker) Del(uniqueKey string, jobType string) error { func (w *worker) Del(jobType string, uniqueKey string) error {
redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType) redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
w.redis.Del(w.ctx, redisKey).Result() w.redis.Del(w.ctx, redisKey).Result()