Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 43d2798b41 | |||
| 1beafa934c |
+2
-2
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user