Compare commits
6 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c3121f425e | |||
| 170275be3b | |||
| 88056ee8e9 | |||
| 4d07ce2c09 | |||
| 43d2798b41 | |||
| 1beafa934c |
+153
-42
@@ -4,8 +4,6 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
|
||||||
"log"
|
|
||||||
"runtime/debug"
|
"runtime/debug"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
@@ -15,6 +13,7 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// 功能描述
|
// 功能描述
|
||||||
|
|
||||||
// 这是基于Redis的定时任务调度器,能够有效的在服务集群里面调度任务,避免了单点压力过高或单点故障问题
|
// 这是基于Redis的定时任务调度器,能够有效的在服务集群里面调度任务,避免了单点压力过高或单点故障问题
|
||||||
// 由于所有的服务代码是一致的,也就是一个定时任务将在所有的服务都有注册,具体调度到哪个服务运行看调度结果
|
// 由于所有的服务代码是一致的,也就是一个定时任务将在所有的服务都有注册,具体调度到哪个服务运行看调度结果
|
||||||
|
|
||||||
@@ -29,23 +28,31 @@ var clusterWorkerList sync.Map
|
|||||||
type Cluster struct {
|
type Cluster struct {
|
||||||
ctx context.Context
|
ctx context.Context
|
||||||
redis *redis.Client
|
redis *redis.Client
|
||||||
|
logger Logger
|
||||||
lockKey string // 全局计算的key
|
lockKey string // 全局计算的key
|
||||||
nextKey string // 下一次执行的key
|
nextKey string // 下一次执行的key
|
||||||
zsetKey string // 有序集合的key
|
zsetKey string // 有序集合的key
|
||||||
listKey string // 可执行的任务列表的key
|
listKey string // 可执行的任务列表的key
|
||||||
|
setKey string // 重入集合的key
|
||||||
}
|
}
|
||||||
|
|
||||||
var clu *Cluster = nil
|
var clu *Cluster = nil
|
||||||
|
|
||||||
func InitCluster(ctx context.Context, red *redis.Client) *Cluster {
|
// 初始化定时器
|
||||||
|
// 全局只需要初始化一次
|
||||||
|
func InitCluster(ctx context.Context, red *redis.Client, keyPrefix string, opts ...Option) *Cluster {
|
||||||
|
op := newOptions(opts...)
|
||||||
|
|
||||||
clusterOnceLimit.Do(func() {
|
clusterOnceLimit.Do(func() {
|
||||||
clu = &Cluster{
|
clu = &Cluster{
|
||||||
ctx: ctx,
|
ctx: ctx,
|
||||||
redis: red,
|
redis: red,
|
||||||
lockKey: "timer:cluster_globalLockKey", // 定时器的全局锁
|
logger: op.logger,
|
||||||
nextKey: "timer:cluster_nextKey",
|
lockKey: keyPrefix + "timer:cluster_globalLockKey", // 定时器的全局锁
|
||||||
zsetKey: "timer:cluster_zsetKey",
|
nextKey: keyPrefix + "timer:cluster_nextKey", // 下一次
|
||||||
listKey: "timer:cluster_listKey",
|
zsetKey: keyPrefix + "timer:cluster_zsetKey", // 有序集合
|
||||||
|
listKey: keyPrefix + "timer:cluster_listKey", // 列表
|
||||||
|
setKey: keyPrefix + "timer:cluster_setKey", // 重入集合
|
||||||
}
|
}
|
||||||
|
|
||||||
// 监听任务
|
// 监听任务
|
||||||
@@ -69,38 +76,108 @@ func InitCluster(ctx context.Context, red *redis.Client) *Cluster {
|
|||||||
return clu
|
return clu
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Duration, callback callback, extendData interface{}) error {
|
// TODO:指定执行时间
|
||||||
_, ok := clusterWorkerList.Load(uniqueKey)
|
// 1.每月的1号2点执行(如果当月没有这个号就不执行)
|
||||||
|
// 2.每周的周一2点执行
|
||||||
|
// 3.每天的2点执行
|
||||||
|
// 4.每小时的2分执行
|
||||||
|
// 5.每分钟的2秒执行
|
||||||
|
func (c *Cluster) AddEveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback callback, extendData interface{}) error {
|
||||||
|
nowTime := time.Now()
|
||||||
|
// 计算下一次执行的时间
|
||||||
|
nextTime := time.Date(nowTime.Year(), nowTime.Month(), day, hour, minute, second, 0, nowTime.Location())
|
||||||
|
if nextTime.Before(nowTime) {
|
||||||
|
nextTime = nextTime.AddDate(0, 1, 0)
|
||||||
|
}
|
||||||
|
return c.addJob(ctx, taskId, nextTime, time.Hour*24*30, callback, extendData, JobTypeEveryMonth, &JobData{Day: &day, Hour: &hour, Minute: &minute, Second: &second})
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Cluster) AddEveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback callback, extendData interface{}) error {
|
||||||
|
nowTime := time.Now()
|
||||||
|
// 计算下一次执行的时间
|
||||||
|
nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), hour, minute, second, 0, nowTime.Location())
|
||||||
|
for nextTime.Weekday() != week {
|
||||||
|
nextTime = nextTime.AddDate(0, 0, 1)
|
||||||
|
}
|
||||||
|
if nextTime.Before(nowTime) {
|
||||||
|
nextTime = nextTime.AddDate(0, 0, 7)
|
||||||
|
}
|
||||||
|
return c.addJob(ctx, taskId, nextTime, time.Hour*24*7, callback, extendData, JobTypeInterval, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Cluster) AddEveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback callback, extendData interface{}) error {
|
||||||
|
nowTime := time.Now()
|
||||||
|
// 计算下一次执行的时间
|
||||||
|
nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), hour, minute, second, 0, nowTime.Location())
|
||||||
|
if nextTime.Before(nowTime) {
|
||||||
|
nextTime = nextTime.AddDate(0, 0, 1)
|
||||||
|
}
|
||||||
|
return c.addJob(ctx, taskId, nextTime, time.Hour*24, callback, extendData, JobTypeInterval, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Cluster) AddEveryHour(ctx context.Context, taskId string, minute int, second int, callback callback, extendData interface{}) error {
|
||||||
|
nowTime := time.Now()
|
||||||
|
// 计算下一次执行的时间
|
||||||
|
nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), nowTime.Hour(), minute, second, 0, nowTime.Location())
|
||||||
|
if nextTime.Before(nowTime) {
|
||||||
|
nextTime = nextTime.Add(time.Hour)
|
||||||
|
}
|
||||||
|
return c.addJob(ctx, taskId, nextTime, time.Hour, callback, extendData, JobTypeInterval, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Cluster) AddEveryMinute(ctx context.Context, taskId string, second int, callback callback, extendData interface{}) error {
|
||||||
|
nowTime := time.Now()
|
||||||
|
// 计算下一次执行的时间
|
||||||
|
nextTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), nowTime.Hour(), nowTime.Minute(), second, 0, nowTime.Location())
|
||||||
|
if nextTime.Before(nowTime) {
|
||||||
|
nextTime = nextTime.Add(time.Minute)
|
||||||
|
}
|
||||||
|
return c.addJob(ctx, taskId, nextTime, time.Minute, callback, extendData, JobTypeInterval, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Cluster) Add(ctx context.Context, taskId string, spaceTime time.Duration, callback callback, extendData interface{}) error {
|
||||||
|
return c.addJob(ctx, taskId, time.Now(), spaceTime, callback, extendData, JobTypeInterval, nil)
|
||||||
|
}
|
||||||
|
|
||||||
|
// 指定时间间隔
|
||||||
|
// TODO:
|
||||||
|
// 1.不同服务定的时间间隔不一致问题
|
||||||
|
// 2.后起的服务计算了时间覆盖前面原有的时间问题
|
||||||
|
func (c *Cluster) addJob(ctx context.Context, taskId string, beginTime time.Time, spaceTime time.Duration, callback callback, extendData interface{}, jobType JobType, jobData *JobData) error {
|
||||||
|
_, ok := clusterWorkerList.Load(taskId)
|
||||||
if ok {
|
if ok {
|
||||||
|
c.logger.Errorf(ctx, "key已存在:%s", taskId)
|
||||||
return errors.New("key已存在")
|
return errors.New("key已存在")
|
||||||
}
|
}
|
||||||
|
|
||||||
if spaceTime != spaceTime.Abs() {
|
if spaceTime != spaceTime.Abs() {
|
||||||
|
c.logger.Errorf(ctx, "时间间隔不能为负数:%s", taskId)
|
||||||
return errors.New("时间间隔不能为负数")
|
return errors.New("时间间隔不能为负数")
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(ctx)
|
ctx, cancel := context.WithCancel(ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
lock := lockx.NewGlobalLock(ctx, c.redis, uniqueKey)
|
lock := lockx.NewGlobalLock(ctx, c.redis, taskId)
|
||||||
tB := lock.Try(10)
|
tB := lock.Try(10)
|
||||||
if !tB {
|
if !tB {
|
||||||
|
c.logger.Errorf(ctx, "添加失败:%s", taskId)
|
||||||
return errors.New("添加失败")
|
return errors.New("添加失败")
|
||||||
}
|
}
|
||||||
defer lock.Unlock()
|
defer lock.Unlock()
|
||||||
|
|
||||||
nowTime := time.Now()
|
|
||||||
|
|
||||||
t := timerStr{
|
t := timerStr{
|
||||||
BeginTime: nowTime,
|
BeginTime: beginTime,
|
||||||
NextTime: nowTime,
|
NextTime: beginTime,
|
||||||
SpaceTime: spaceTime,
|
SpaceTime: spaceTime,
|
||||||
Callback: callback,
|
Callback: callback,
|
||||||
ExtendData: extendData,
|
ExtendData: extendData,
|
||||||
UniqueKey: uniqueKey,
|
TaskId: taskId,
|
||||||
|
JobType: jobType,
|
||||||
|
JobData: jobData,
|
||||||
}
|
}
|
||||||
|
|
||||||
clusterWorkerList.Store(uniqueKey, t)
|
clusterWorkerList.Store(taskId, t)
|
||||||
|
|
||||||
cacheStr, _ := c.redis.Get(ctx, c.nextKey).Result()
|
cacheStr, _ := c.redis.Get(ctx, c.nextKey).Result()
|
||||||
execTime := make(map[string]time.Time)
|
execTime := make(map[string]time.Time)
|
||||||
@@ -110,9 +187,9 @@ func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Dura
|
|||||||
|
|
||||||
p.ZAdd(ctx, c.zsetKey, &redis.Z{
|
p.ZAdd(ctx, c.zsetKey, &redis.Z{
|
||||||
Score: float64(nextTime.UnixMilli()),
|
Score: float64(nextTime.UnixMilli()),
|
||||||
Member: uniqueKey,
|
Member: taskId,
|
||||||
})
|
})
|
||||||
execTime[uniqueKey] = nextTime
|
execTime[taskId] = nextTime
|
||||||
n, _ := json.Marshal(execTime)
|
n, _ := json.Marshal(execTime)
|
||||||
// fmt.Println("execTime:", execTime, string(n))
|
// fmt.Println("execTime:", execTime, string(n))
|
||||||
p.Set(ctx, c.nextKey, string(n), 0)
|
p.Set(ctx, c.nextKey, string(n), 0)
|
||||||
@@ -125,9 +202,9 @@ func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Dura
|
|||||||
}
|
}
|
||||||
|
|
||||||
// 计算下一次执行的时间
|
// 计算下一次执行的时间
|
||||||
|
// TODO:注册的任务需放在Redis集中存储,因为本地的话,如果有多个服务,那么就会出现不一致的情况。但是要注意服务如何进行下线,由于是主动上报的,需要有一个机制进行删除过期的任务(添加任务&定时器轮训注册)
|
||||||
func (c *Cluster) getNextTime() {
|
func (c *Cluster) getNextTime() {
|
||||||
|
|
||||||
// log.Println("begin computer")
|
|
||||||
ctx, cancel := context.WithCancel(c.ctx)
|
ctx, cancel := context.WithCancel(c.ctx)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
@@ -153,18 +230,18 @@ func (c *Cluster) getNextTime() {
|
|||||||
|
|
||||||
clusterWorkerList.Range(func(key, value interface{}) bool {
|
clusterWorkerList.Range(func(key, value interface{}) bool {
|
||||||
val := value.(timerStr)
|
val := value.(timerStr)
|
||||||
beforeTime := execTime[val.UniqueKey]
|
beforeTime := execTime[val.TaskId]
|
||||||
if beforeTime.After(nowTime) {
|
if beforeTime.After(nowTime) {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
nextTime := getNextExecTime(beforeTime, val.SpaceTime)
|
nextTime := getNextExecTime(val)
|
||||||
execTime[val.UniqueKey] = nextTime
|
execTime[val.TaskId] = nextTime
|
||||||
|
|
||||||
p.ZAdd(ctx, c.zsetKey, &redis.Z{
|
p.ZAdd(ctx, c.zsetKey, &redis.Z{
|
||||||
Score: float64(nextTime.UnixMilli()),
|
Score: float64(nextTime.UnixMilli()),
|
||||||
Member: val.UniqueKey,
|
Member: val.TaskId,
|
||||||
})
|
})
|
||||||
// log.Println("computeTime add", c.zsetKey, val.UniqueKey, nextTime.UnixMilli())
|
// log.Println("computeTime add", c.zsetKey, val.taskId, nextTime.UnixMilli())
|
||||||
return true
|
return true
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -177,14 +254,16 @@ func (c *Cluster) getNextTime() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// 递归遍历获取执行时间
|
// 递归遍历获取执行时间
|
||||||
func getNextExecTime(beforeTime time.Time, spaceTime time.Duration) time.Time {
|
// TODO:需要根据不同的任务类型计算下次定时时间
|
||||||
|
func getNextExecTime(ts timerStr) time.Time {
|
||||||
nowTime := time.Now()
|
nowTime := time.Now()
|
||||||
if beforeTime.After(nowTime) {
|
if ts.NextTime.After(nowTime) {
|
||||||
return beforeTime
|
return ts.NextTime
|
||||||
}
|
}
|
||||||
nextTime := beforeTime.Add(spaceTime)
|
nextTime := ts.NextTime.Add(ts.SpaceTime)
|
||||||
|
ts.NextTime = nextTime
|
||||||
if nextTime.Before(nowTime) {
|
if nextTime.Before(nowTime) {
|
||||||
nextTime = getNextExecTime(nextTime, spaceTime)
|
nextTime = getNextExecTime(ts)
|
||||||
}
|
}
|
||||||
return nextTime
|
return nextTime
|
||||||
}
|
}
|
||||||
@@ -207,29 +286,61 @@ func (c *Cluster) getTask() {
|
|||||||
// 监听任务
|
// 监听任务
|
||||||
func (c *Cluster) watch() {
|
func (c *Cluster) watch() {
|
||||||
// 执行任务
|
// 执行任务
|
||||||
for {
|
go func() {
|
||||||
keys, err := c.redis.BLPop(c.ctx, time.Second*10, c.listKey).Result()
|
for {
|
||||||
if err != nil {
|
keys, err := c.redis.BLPop(c.ctx, time.Second*10, c.listKey).Result()
|
||||||
fmt.Println("watch err:", err)
|
if err != nil {
|
||||||
continue
|
if err != redis.Nil {
|
||||||
|
c.logger.Errorf(c.ctx, "BLPop watch err:%+v", err)
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_, ok := clusterWorkerList.Load(keys[1])
|
||||||
|
if !ok {
|
||||||
|
c.logger.Errorf(c.ctx, "watch timer:任务不存在", keys[1])
|
||||||
|
c.redis.SAdd(c.ctx, c.setKey, keys[1])
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
go c.doTask(c.ctx, c.redis, keys[1])
|
||||||
}
|
}
|
||||||
go doTask(c.ctx, c.redis, keys[1])
|
}()
|
||||||
}
|
|
||||||
|
go func() {
|
||||||
|
for {
|
||||||
|
taskId, err := c.redis.SPop(c.ctx, c.setKey).Result()
|
||||||
|
if err != nil {
|
||||||
|
if err == redis.Nil {
|
||||||
|
// 已经是空了就不要浪费资源了
|
||||||
|
time.Sleep(time.Second)
|
||||||
|
} else {
|
||||||
|
c.logger.Errorf(c.ctx, "SPop watch err:%+v", err)
|
||||||
|
}
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
_, ok := clusterWorkerList.Load(taskId)
|
||||||
|
if !ok {
|
||||||
|
c.logger.Errorf(c.ctx, "watch timer:任务不存在", taskId)
|
||||||
|
c.redis.SAdd(c.ctx, c.setKey, taskId)
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
go c.doTask(c.ctx, c.redis, taskId)
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// 执行任务
|
// 执行任务
|
||||||
func doTask(ctx context.Context, red *redis.Client, taskId string) {
|
func (c *Cluster) doTask(ctx context.Context, red *redis.Client, taskId string) {
|
||||||
|
|
||||||
defer func() {
|
defer func() {
|
||||||
if err := recover(); err != nil {
|
if err := recover(); err != nil {
|
||||||
fmt.Println("timer:定时器出错", err)
|
c.logger.Errorf(ctx, "timer:定时器出错 err:%+v stack:%s", err, string(debug.Stack()))
|
||||||
log.Println("errStack", string(debug.Stack()))
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
||||||
val, ok := clusterWorkerList.Load(taskId)
|
val, ok := clusterWorkerList.Load(taskId)
|
||||||
if !ok {
|
if !ok {
|
||||||
fmt.Println("doTask timer:任务不存在", taskId)
|
c.logger.Errorf(ctx, "doTask timer:任务不存在", taskId)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
t := val.(timerStr)
|
t := val.(timerStr)
|
||||||
@@ -238,11 +349,11 @@ func doTask(ctx context.Context, red *redis.Client, taskId string) {
|
|||||||
lock := lockx.NewGlobalLock(ctx, red, taskId)
|
lock := lockx.NewGlobalLock(ctx, red, taskId)
|
||||||
tB := lock.Lock()
|
tB := lock.Lock()
|
||||||
if !tB {
|
if !tB {
|
||||||
fmt.Println("doTask timer:获取锁失败", taskId)
|
c.logger.Errorf(ctx, "doTask timer:获取锁失败", taskId)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
defer lock.Unlock()
|
defer lock.Unlock()
|
||||||
|
|
||||||
// 执行任务
|
// 执行任务
|
||||||
t.Callback(ctx,t.ExtendData)
|
t.Callback(ctx, t.ExtendData)
|
||||||
}
|
}
|
||||||
|
|||||||
+4
-4
@@ -30,7 +30,7 @@ func main() {
|
|||||||
|
|
||||||
func worker() {
|
func worker() {
|
||||||
client := getRedis()
|
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{}{
|
w.Add("test", "test", 1*time.Second, map[string]interface{}{
|
||||||
"test": "test",
|
"test": "test",
|
||||||
})
|
})
|
||||||
@@ -52,11 +52,11 @@ func worker() {
|
|||||||
|
|
||||||
type Worker struct{}
|
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("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
|
||||||
fmt.Println(uniqueKey, jobType)
|
fmt.Println(uniqueKey, jobType)
|
||||||
fmt.Println(data)
|
fmt.Println(data)
|
||||||
return timerx.WorkerCodeAgain,time.Second
|
return timerx.WorkerCodeAgain, time.Second
|
||||||
}
|
}
|
||||||
|
|
||||||
func getRedis() *redis.Client {
|
func getRedis() *redis.Client {
|
||||||
@@ -76,7 +76,7 @@ func re() {
|
|||||||
client := getRedis()
|
client := getRedis()
|
||||||
|
|
||||||
ctx := context.Background()
|
ctx := context.Background()
|
||||||
cl := timerx.InitCluster(ctx, client)
|
cl := timerx.InitCluster(ctx, client, "kkkk")
|
||||||
cl.Add(ctx, "test1", 1*time.Millisecond, aa, "data")
|
cl.Add(ctx, "test1", 1*time.Millisecond, aa, "data")
|
||||||
cl.Add(ctx, "test2", 1*time.Millisecond, aa, "data")
|
cl.Add(ctx, "test2", 1*time.Millisecond, aa, "data")
|
||||||
cl.Add(ctx, "test3", 1*time.Millisecond, aa, "data")
|
cl.Add(ctx, "test3", 1*time.Millisecond, aa, "data")
|
||||||
|
|||||||
@@ -1,11 +1,13 @@
|
|||||||
module code.yun.ink/pkg/timerx
|
module github.com/yuninks/timerx
|
||||||
|
|
||||||
go 1.19
|
go 1.19
|
||||||
|
|
||||||
require github.com/go-redis/redis/v8 v8.11.5
|
require (
|
||||||
|
code.yun.ink/pkg/lockx v1.0.0
|
||||||
|
github.com/go-redis/redis/v8 v8.11.5
|
||||||
|
)
|
||||||
|
|
||||||
require (
|
require (
|
||||||
code.yun.ink/pkg/lockx v1.0.0 // indirect
|
|
||||||
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -1,3 +1,4 @@
|
|||||||
|
code.yun.ink/open/timer v1.0.1 h1:ZWecU5K0rFB15p8DZubozTEwo1vrO4mUCRwEoD1tbEQ=
|
||||||
code.yun.ink/pkg/lockx v1.0.0 h1:xoLyf05PrOAhLID2LbJsEXA8YYURJTK/7spEk/hu/Rs=
|
code.yun.ink/pkg/lockx v1.0.0 h1:xoLyf05PrOAhLID2LbJsEXA8YYURJTK/7spEk/hu/Rs=
|
||||||
code.yun.ink/pkg/lockx v1.0.0/go.mod h1:0xUU5xD8fui0Kf7g4TnFmaxUDo59CH2WM+sitko2SLc=
|
code.yun.ink/pkg/lockx v1.0.0/go.mod h1:0xUU5xD8fui0Kf7g4TnFmaxUDo59CH2WM+sitko2SLc=
|
||||||
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
|
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
|
||||||
|
|||||||
@@ -0,0 +1,25 @@
|
|||||||
|
package timerx
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"log"
|
||||||
|
)
|
||||||
|
|
||||||
|
type Logger interface {
|
||||||
|
Infof(ctx context.Context, format string, v ...interface{})
|
||||||
|
Errorf(ctx context.Context, format string, v ...interface{})
|
||||||
|
}
|
||||||
|
|
||||||
|
type defaultLogger struct{}
|
||||||
|
|
||||||
|
func NewLogger() *defaultLogger {
|
||||||
|
return &defaultLogger{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *defaultLogger) Infof(ctx context.Context, format string, v ...interface{}) {
|
||||||
|
log.Printf("[INFO] "+format, v...)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *defaultLogger) Errorf(ctx context.Context, format string, v ...interface{}) {
|
||||||
|
log.Printf("[ERROR] "+format, v...)
|
||||||
|
}
|
||||||
@@ -14,9 +14,9 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// 功能描述
|
// 功能描述
|
||||||
// 1. 任务全局唯一
|
// 1. 任务可以多节点发布
|
||||||
// 2. 任务只执行一次
|
// 2. 每个任务的执行在全局仅会执行一次
|
||||||
// 3. 任务执行失败可以重新放入队列
|
// 3. 任务执行失败支持快捷重新加入队列
|
||||||
|
|
||||||
// 单次的任务队列
|
// 单次的任务队列
|
||||||
type worker struct {
|
type worker struct {
|
||||||
@@ -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()
|
||||||
|
|
||||||
|
|||||||
@@ -0,0 +1,35 @@
|
|||||||
|
package timerx
|
||||||
|
|
||||||
|
type Options struct {
|
||||||
|
logger Logger
|
||||||
|
}
|
||||||
|
|
||||||
|
func defaultOptions() Options {
|
||||||
|
return Options{
|
||||||
|
logger: NewLogger(),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
type Option func(*Options)
|
||||||
|
|
||||||
|
func newOptions(opts ...Option) Options {
|
||||||
|
o := defaultOptions()
|
||||||
|
for _, opt := range opts {
|
||||||
|
opt(&o)
|
||||||
|
}
|
||||||
|
return o
|
||||||
|
}
|
||||||
|
|
||||||
|
// 设置日志
|
||||||
|
func SetLogger(log Logger) Option {
|
||||||
|
return func(o *Options) {
|
||||||
|
o.logger = log
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// 设定时区
|
||||||
|
func SetTimeZone(zone string) Option {
|
||||||
|
return func(o *Options) {
|
||||||
|
// todo
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -6,15 +6,13 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"log"
|
|
||||||
"runtime/debug"
|
"runtime/debug"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
// 定时器
|
// 定时器
|
||||||
// 原理:每毫秒的时间触发
|
// 1. 这个定时器的作用范围是本机
|
||||||
// 单机版重复时间间隔定时器
|
|
||||||
|
|
||||||
// uuid -> timerStr
|
// uuid -> timerStr
|
||||||
var timerMap = make(map[string]*timerStr)
|
var timerMap = make(map[string]*timerStr)
|
||||||
@@ -23,14 +21,23 @@ var timerMapMux sync.Mutex
|
|||||||
var timerCount int // 当前定时数目
|
var timerCount int // 当前定时数目
|
||||||
var onceLimit sync.Once // 实现单例
|
var onceLimit sync.Once // 实现单例
|
||||||
|
|
||||||
type Single struct{}
|
type Single struct {
|
||||||
|
ctx context.Context
|
||||||
|
logger Logger
|
||||||
|
}
|
||||||
|
|
||||||
var sin *Single = nil
|
var sin *Single = nil
|
||||||
|
|
||||||
// 定时器类
|
// 定时器类
|
||||||
func InitSingle(ctx context.Context) *Single {
|
func InitSingle(ctx context.Context, opts ...Option) *Single {
|
||||||
onceLimit.Do(func() {
|
onceLimit.Do(func() {
|
||||||
sin = &Single{}
|
|
||||||
|
op := newOptions(opts...)
|
||||||
|
|
||||||
|
sin = &Single{
|
||||||
|
ctx: ctx,
|
||||||
|
logger: op.logger,
|
||||||
|
}
|
||||||
|
|
||||||
timer := time.NewTicker(time.Millisecond * 200)
|
timer := time.NewTicker(time.Millisecond * 200)
|
||||||
go func(ctx context.Context) {
|
go func(ctx context.Context) {
|
||||||
@@ -50,7 +57,7 @@ func InitSingle(ctx context.Context) *Single {
|
|||||||
break Loop
|
break Loop
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
log.Println("timer: initend")
|
sin.logger.Infof(ctx, "timer: initend")
|
||||||
}(ctx)
|
}(ctx)
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -68,6 +75,7 @@ func (s *Single) Add(space time.Duration, call callback, extend interface{}) (in
|
|||||||
defer timerMapMux.Unlock()
|
defer timerMapMux.Unlock()
|
||||||
|
|
||||||
if space != space.Abs() {
|
if space != space.Abs() {
|
||||||
|
s.logger.Errorf(s.ctx, "space must be positive")
|
||||||
return 0, errors.New("space must be positive")
|
return 0, errors.New("space must be positive")
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -81,7 +89,6 @@ func (s *Single) Add(space time.Duration, call callback, extend interface{}) (in
|
|||||||
NextTime: nowTime, // nowTime.Add(space), // 添加任务的时候就执行一次
|
NextTime: nowTime, // nowTime.Add(space), // 添加任务的时候就执行一次
|
||||||
SpaceTime: space,
|
SpaceTime: space,
|
||||||
CanRunning: make(chan struct{}, 1),
|
CanRunning: make(chan struct{}, 1),
|
||||||
UniqueKey: "",
|
|
||||||
ExtendData: extend,
|
ExtendData: extend,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -152,7 +159,7 @@ func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
|
|||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
// fmt.Printf("timer: 准备执行 %v %v \n", k, v.Tag)
|
// fmt.Printf("timer: 准备执行 %v %v \n", k, v.Tag)
|
||||||
s.doTask(ctx, v.Callback, v.UniqueKey, v.ExtendData)
|
s.doTask(ctx, v.Callback, v.ExtendData)
|
||||||
default:
|
default:
|
||||||
// fmt.Printf("timer: 已在执行 %v %v \n", k, v.Tag)
|
// fmt.Printf("timer: 已在执行 %v %v \n", k, v.Tag)
|
||||||
return
|
return
|
||||||
@@ -177,11 +184,10 @@ func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
|
|||||||
|
|
||||||
// 定时器操作类
|
// 定时器操作类
|
||||||
// 这里不应painc
|
// 这里不应painc
|
||||||
func (s *Single) doTask(ctx context.Context, call callback, uniqueKey string, extend interface{}) error {
|
func (s *Single) doTask(ctx context.Context, call callback, extend interface{}) error {
|
||||||
defer func() {
|
defer func() {
|
||||||
if err := recover(); err != nil {
|
if err := recover(); err != nil {
|
||||||
fmt.Println("timer:定时器出错", err)
|
s.logger.Errorf(ctx, "timer:定时器出错 err:%+v stack:%s", err, string(debug.Stack()))
|
||||||
log.Println("errStack", string(debug.Stack()))
|
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
return call(ctx, extend)
|
return call(ctx, extend)
|
||||||
|
|||||||
+9
-4
@@ -1,15 +1,20 @@
|
|||||||
package timerx_test
|
package timerx_test
|
||||||
|
|
||||||
import "testing"
|
import (
|
||||||
|
"fmt"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
// 单元测试
|
// 单元测试
|
||||||
|
|
||||||
func TestHelloWorld(t *testing.T) {
|
func TestHelloWorld(t *testing.T) {
|
||||||
// 日志
|
// 日志
|
||||||
t.Log("hello world")
|
// t.Log("hello world")
|
||||||
|
|
||||||
s := "ddd"
|
fmt.Println("hello world")
|
||||||
t.Logf("Log测试%s", s)
|
|
||||||
|
// s := "ddd"
|
||||||
|
// t.Logf("Log测试%s", s)
|
||||||
// t.Errorf("ErrorF %s", s)
|
// t.Errorf("ErrorF %s", s)
|
||||||
|
|
||||||
// 标记错误(继续运行)
|
// 标记错误(继续运行)
|
||||||
|
|||||||
@@ -11,13 +11,34 @@ type timerStr struct {
|
|||||||
BeginTime time.Time // 初始化任务的时间
|
BeginTime time.Time // 初始化任务的时间
|
||||||
NextTime time.Time // [删]下一次执行的时间
|
NextTime time.Time // [删]下一次执行的时间
|
||||||
SpaceTime time.Duration // 任务间隔时间
|
SpaceTime time.Duration // 任务间隔时间
|
||||||
UniqueKey string // 全局唯一键
|
TaskId string // 任务ID 全局唯一键
|
||||||
ExtendData interface{} // 附加参数
|
ExtendData interface{} // 附加参数
|
||||||
|
JobType JobType // 任务类型
|
||||||
|
JobData *JobData // 任务时间数据
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type JobType string
|
||||||
|
|
||||||
|
const (
|
||||||
|
JobTypeEveryDay JobType = "every_day"
|
||||||
|
JobTypeEveryHour JobType = "every_hour"
|
||||||
|
JobTypeEveryMinute JobType = "every_minute"
|
||||||
|
JobTypeEverySecond JobType = "every_second"
|
||||||
|
JobTypeEveryMonth JobType = "every_month"
|
||||||
|
// 根据间隔时间执行
|
||||||
|
JobTypeInterval JobType = "interval"
|
||||||
|
)
|
||||||
|
|
||||||
|
type JobData struct {
|
||||||
|
Month *time.Month // 每年的第几个月
|
||||||
|
Weekday *time.Weekday // 每周的周几
|
||||||
|
Day *int // 每月的第几天
|
||||||
|
Hour *int // 每天的第几个小时
|
||||||
|
Minute *int // 每小时的第几分钟
|
||||||
|
Second *int // 每分钟的第几秒
|
||||||
|
}
|
||||||
|
|
||||||
var nextTime = time.Now() // 下一次执行的时间
|
var nextTime = time.Now() // 下一次执行的时间
|
||||||
|
|
||||||
|
|
||||||
// 定义各个回调函数
|
// 定义各个回调函数
|
||||||
type callback func(ctx context.Context, extendData interface{}) error
|
type callback func(ctx context.Context, extendData interface{}) error
|
||||||
|
|||||||
Reference in New Issue
Block a user