2 Commits

Author SHA1 Message Date
yun 43d2798b41 优化定时器的表述 2023-12-27 17:19:52 +08:00
yun 1beafa934c 优化全局单次定时器的冲突问题 2023-12-27 17:11:33 +08:00
3 changed files with 15 additions and 16 deletions
+2 -2
View File
@@ -30,7 +30,7 @@ func main() {
func worker() {
client := getRedis()
w := timerx.InitOnce(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,7 +52,7 @@ func worker() {
type Worker struct{}
func (w *Worker) Worker(uniqueKey string, jobType string, data interface{}) (timerx.WorkerCode, time.Duration) {
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)
+12 -12
View File
@@ -14,9 +14,9 @@ import (
)
// 功能描述
// 1. 任务全局唯一
// 2. 任务只执行一次
// 3. 任务执行失败可以重新入队列
// 1. 任务可以多节点发布
// 2. 每个任务的执行在全局仅会执行一次
// 3. 任务执行失败支持快捷重新入队列
// 单次的任务队列
type worker struct {
@@ -40,7 +40,7 @@ type Callback interface {
// uniqueKey: 任务唯一标识
// jobType: 任务类型,用于区分任务
// 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
@@ -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() {
wo = &worker{
ctx: ctx,
zsetKey: "timer:once_zsetkey",
listKey: "timer:once_listkey",
zsetKey: "timer:once_zsetkey" + jobGlobalName,
listKey: "timer:once_listkey" + jobGlobalName,
redis: re,
worker: w,
worker: jobCallback,
}
go wo.getTask()
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 {
return fmt.Errorf("时间间隔不能为负数")
}
@@ -79,7 +79,7 @@ func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration,
return fmt.Errorf("时间间隔不能为0")
}
redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType)
redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
ed := extendData{
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 {
redisKey := fmt.Sprintf("%s[:]%s", uniqueKey, jobType)
func (w *worker) Del(jobType string, uniqueKey string) error {
redisKey := fmt.Sprintf("%s[:]%s", jobType, uniqueKey)
w.redis.Del(w.ctx, redisKey).Result()
+1 -2
View File
@@ -13,8 +13,7 @@ import (
)
// 定时器
// 原理:每毫秒的时间触发
// 单机版重复时间间隔定时器
// 1. 这个定时器的作用范围是本机
// uuid -> timerStr
var timerMap = make(map[string]*timerStr)