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

优化部分存在的问题

Browse Source
This commit is contained in:
Yun
2026-05-27 10:56:37 +08:00
parent fa359b2b1a
commit 448116dc0d
11 changed files with 109 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cluster.go
+26 -18
View File
@@ -70,7 +70,11 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
ctx, cancel := context.WithCancel(ctx)
U, _ := uuid.NewV7()
U, err := uuid.NewV7()
if err != nil {
op.logger.Errorf(ctx, "InitCluster uuid.NewV7 err:%v, using fallback", err)
U = uuid.New()
}
clu := &Cluster{
ctx: ctx,
@@ -204,7 +208,7 @@ func (l *Cluster) startDaemon() {
}
func (l *Cluster) cleanExecuteInfoLoop() {
l.wg.Done()
defer l.wg.Done()
ticker := time.NewTicker(time.Minute * 5)
defer ticker.Stop()
@@ -224,10 +228,11 @@ func (l *Cluster) cleanExecuteInfoLoop() {
}
// 清除过期任务
func (l *Cluster) cleanExecuteInfo() error {
func (l *Cluster) cleanExecuteInfo() {
// 移除执行信息
l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err()
return nil
if err := l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err(); err != nil {
l.logger.Errorf(l.ctx, "cleanExecuteInfo ZRemRangeByScore err:%v", err)
}
}
// scheduleTasks 调度任务（只有leader执行）
@@ -449,7 +454,11 @@ func (l *Cluster) calculateNextTimes() {
// 根据内部注册的任务列表计算下一次执行的时间
l.workerList.Range(func(key, value interface{}) bool {
val := value.(timerStr)
val, ok := value.(timerStr)
if !ok {
l.logger.Errorf(l.ctx, "Cluster calculateNextTimes invalid type in workerList for key:%v", key)
return true
}
nextTime, err := GetNextTime(time.Now().In(l.location), *val.JobData)
if err != nil {
@@ -539,7 +548,7 @@ func (c *Cluster) executeTasks() {
case <-c.ctx.Done():
return
case c.workerChan <- struct{}{}:
func() {
go func() {
defer func() {
<-c.workerChan
}()
@@ -553,7 +562,6 @@ func (c *Cluster) executeTasks() {
if err != nil {
if err != redis.Nil {
c.logger.Errorf(c.ctx, "Failed to pop task: %v", err)
// Redis 异常，休眠一会儿
time.Sleep(5 * time.Second)
}
return
@@ -561,22 +569,16 @@ func (c *Cluster) executeTasks() {
if len(taskID) < 2 {
c.logger.Errorf(c.ctx, "Invalid BLPop result: %v", taskID)
// 数据异常，继续下一个
return
}
go c.processTask(taskID[1])
c.processTask(taskID[1])
}()
}
}
}
type ReJobData struct {
TaskId string
Times int
}
// 执行任务
func (l *Cluster) processTask(taskId string) {
@@ -585,7 +587,11 @@ func (l *Cluster) processTask(taskId string) {
ctx, cancel := context.WithTimeout(l.ctx, l.timeout)
defer cancel()
u, _ := uuid.NewV7()
u, err := uuid.NewV7()
if err != nil {
l.logger.Errorf(ctx, "processTask uuid.NewV7 err:%v, using fallback", err)
u = uuid.New()
}
ctx = context.WithValue(ctx, "trace_id", u.String())
@@ -593,10 +599,12 @@ func (l *Cluster) processTask(taskId string) {
// 上报执行情况
executeVal := fmt.Sprintf("tid:%s|insId:%s|uuid:%s|time:%s", taskId, l.instanceId, u.String(), begin.Format(time.RFC3339Nano))
l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{
if err := l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{
Score: float64(begin.UnixMilli()),
Member: executeVal,
})
}).Err(); err != nil {
l.logger.Errorf(ctx, "processTask ZAdd executeInfo err:%v", err)
}
val, ok := l.workerList.Load(taskId)
if !ok {
heartbeat/heartbeat.go
+3 -1
View File
@@ -150,7 +150,9 @@ func (l *HeartBeat) cleanHeartbeat(cleanSelf bool) error {
}
// 移除心跳
l.redis.ZRemRangeByScore(l.ctx, l.heartbeatKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Second).UnixMilli(), 10)).Err()
if err := l.redis.ZRemRangeByScore(l.ctx, l.heartbeatKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Second).UnixMilli(), 10)).Err(); err != nil {
l.logger.Errorf(l.ctx, "cleanHeartbeat ZRemRangeByScore err:%v", err)
}
return nil
}
heartbeat/options.go
+4 -1
View File
@@ -17,7 +17,10 @@ type Options struct {
func defaultOptions() Options {
u, _ := uuid.NewV7()
u, err := uuid.NewV7()
if err != nil {
u = uuid.New()
}
return Options{
logger: logger.NewLogger(),
leader/leader.go
+9 -6
View File
@@ -111,8 +111,11 @@ func (l *Leader) getLeaderLock() error {
l.logger.Errorf(l.ctx, "getLeaderLock err:%+v", err)
return err
}
if b, _ := lock.Lock(); !b {
// 加锁失败 非Reader
if b, err := lock.Lock(); !b || err != nil {
// 加锁失败 非Leader
if err != nil {
l.logger.Errorf(l.ctx, "getLeaderLock lock.Lock err:%+v", err)
}
l.leaderLock.Lock()
l.isLeader = false
l.leaderLock.Unlock()
@@ -152,10 +155,10 @@ func (l *Leader) getLeaderLock() error {
// 等待超时退出
<-lock.GetCtx().Done()
// 已过期
// l.leaderLock.Lock()
// l.isLeader = false
// l.leaderLock.Unlock()
// 锁已过期，重置leader状态
l.leaderLock.Lock()
l.isLeader = false
l.leaderLock.Unlock()
return nil
leader/options.go
+4 -1
View File
@@ -15,7 +15,10 @@ type Options struct {
func defaultOptions() Options {
u, _ := uuid.NewV7()
u, err := uuid.NewV7()
if err != nil {
u = uuid.New()
}
return Options{
logger: logger.NewLogger(),
next_time.go
+3 -41
View File
@@ -240,56 +240,18 @@ func calculateNextCronTime(t time.Time, job JobData) (*time.Time, error) {
return nil, ErrCronParser
}
s := *job.CronSchedule
next := s.Next(t)
next := job.CronSchedule.Next(t)
return &next, nil
}
func GetCronSche(CronExpression string, cronParser *cron.Parser) (*cron.Schedule, error) {
func GetCronSche(CronExpression string, cronParser *cron.Parser) (cron.Schedule, error) {
if CronExpression == "" {
return nil, ErrCronExpression
}
if cronParser == nil {
return nil, ErrCronParser
}
sche, err := cronParser.Parse(CronExpression)
if err != nil {
return nil, err
}
return &sche, nil
return cronParser.Parse(CronExpression)
}
// 检查是否本周期可以运行
// 检查是否本周期可以运行（已弃用，使用新的时间比较逻辑）
// 保留此函数用于向后兼容，但建议使用新的时间计算逻辑
func canRun(t time.Time, job JobData) bool {
targetTime := time.Date(t.Year(), t.Month(), t.Day(), job.Hour, job.Minute, job.Second, 0, t.Location())
switch job.JobType {
case JobTypeEveryMonth:
// 对于月任务，需要比较日期
targetTime = time.Date(t.Year(), t.Month(), job.Day, job.Hour, job.Minute, job.Second, 0, t.Location())
return !targetTime.Before(t)
case JobTypeEveryWeek:
// 对于周任务，需要比较星期
currentWeekday := t.Weekday()
if currentWeekday < job.Weekday {
return true
}
if currentWeekday == job.Weekday {
return targetTime.After(t) || targetTime.Equal(t)
}
return false
case JobTypeEveryDay:
return targetTime.After(t) || targetTime.Equal(t)
case JobTypeEveryHour:
hourTarget := time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), job.Minute, job.Second, 0, t.Location())
return hourTarget.After(t) || hourTarget.Equal(t)
case JobTypeEveryMinute:
minuteTarget := time.Date(t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), job.Second, 0, t.Location())
return minuteTarget.After(t) || minuteTarget.Equal(t)
default:
return false
}
}
once.go
+46 -17
View File
@@ -108,7 +108,11 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
ctx, cancel := context.WithCancel(ctx)
u, _ := uuid.NewV7()
u, err := uuid.NewV7()
if err != nil {
op.logger.Errorf(ctx, "InitOnce uuid.NewV7 err:%v, using fallback", err)
u = uuid.New()
}
wo := &Once{
ctx: ctx,
@@ -233,7 +237,7 @@ func (l *Once) startDaemon() {
}
func (l *Once) cleanExecuteInfoLoop() {
l.wg.Done()
defer l.wg.Done()
ticker := time.NewTicker(time.Minute * 5)
defer ticker.Stop()
@@ -253,10 +257,11 @@ func (l *Once) cleanExecuteInfoLoop() {
}
// 清除过期任务
func (l *Once) cleanExecuteInfo() error {
func (l *Once) cleanExecuteInfo() {
// 移除执行信息
l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err()
return nil
if err := l.redis.ZRemRangeByScore(l.ctx, l.executeInfoKey, "0", strconv.FormatInt(time.Now().Add(-15*time.Minute).UnixMilli(), 10)).Err(); err != nil {
l.logger.Errorf(l.ctx, "cleanExecuteInfo ZRemRangeByScore err:%v", err)
}
}
// 任务调度 领导
@@ -302,7 +307,7 @@ func (l *Once) executeTasks() {
case <-l.ctx.Done():
return
case l.workerChan <- struct{}{}:
func() {
go func() {
defer func() {
<-l.workerChan
}()
@@ -316,7 +321,6 @@ func (l *Once) executeTasks() {
if err != nil {
if err != redis.Nil {
l.logger.Errorf(l.ctx, "Failed to pop task: %v", err)
// Redis 异常，休眠一会儿再重试
time.Sleep(time.Second * 5)
}
return
@@ -324,11 +328,10 @@ func (l *Once) executeTasks() {
if len(keys) < 2 {
l.logger.Errorf(l.ctx, "Invalid task data: %v", keys)
// 数据异常，继续下一个
return
}
// 处理任务
go l.processTask(keys[1])
l.processTask(keys[1])
}()
}
}
@@ -409,7 +412,11 @@ func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType,
RunCount: runCount,
JobType: jobType,
}
b, _ := json.Marshal(ed)
b, err := json.Marshal(ed)
if err != nil {
w.logger.Errorf(ctx, "save json.Marshal err:%v taskType:%v taskId:%v", err, taskType, taskId)
return err
}
// 使用事务确保原子性
pipe := w.redis.TxPipeline()
@@ -424,7 +431,7 @@ func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType,
Score: float64(nextTime.UnixMilli()),
Member: redisKey,
})
_, err := pipe.Exec(w.ctx)
_, err = pipe.Exec(w.ctx)
if err != nil {
w.logger.Errorf(w.ctx, "save task failed:%v taskType:%v taskId:%v attachData:%v retryCount:%v", err, taskType, taskId, attachData, runCount)
return err
@@ -509,8 +516,21 @@ func (w *Once) Delete(taskType OnceTaskType, taskId string) error {
}
// 获取任务
func (l *Once) Get(taskType OnceTaskType, taskId string) {
//
func (l *Once) Get(taskType OnceTaskType, taskId string) (any, error) {
redisKey := l.buildRedisKey(taskType, taskId)
dataKey := fmt.Sprintf("timer:{%s}:once_data:%s", l.keyPrefix, redisKey)
str, err := l.redis.Get(l.ctx, dataKey).Result()
if err != nil {
if errors.Is(err, redis.Nil) {
return nil, nil
}
return nil, err
}
var ed extendData
if err := json.Unmarshal([]byte(str), &ed); err != nil {
return nil, err
}
return ed.Data, nil
}
// 批量获取任务
@@ -551,7 +571,11 @@ func (l *Once) processTask(key string) {
ctx, cancel := context.WithTimeout(l.ctx, l.timeout)
defer cancel()
u, _ := uuid.NewV7()
u, err := uuid.NewV7()
if err != nil {
l.logger.Errorf(ctx, "processTask uuid.NewV7 err:%v, using fallback", err)
u = uuid.New()
}
ctx = context.WithValue(ctx, "trace_id", u.String())
@@ -577,10 +601,12 @@ func (l *Once) processTask(key string) {
// 上报执行情况
executeVal := fmt.Sprintf("tid:%s|insId:%s|uuid:%s|time:%s", key, l.instanceId, u.String(), begin.Format(time.RFC3339Nano))
l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{
if err := l.redis.ZAdd(ctx, l.executeInfoKey, redis.Z{
Score: float64(begin.UnixMilli()),
Member: executeVal,
})
}).Err(); err != nil {
l.logger.Errorf(ctx, "processTask ZAdd executeInfo err:%v", err)
}
defer func() {
if err := recover(); err != nil {
@@ -597,7 +623,10 @@ func (l *Once) processTask(key string) {
}
ed := extendData{}
json.Unmarshal([]byte(str), &ed)
if err := json.Unmarshal([]byte(str), &ed); err != nil {
l.logger.Errorf(ctx, "processTask json.Unmarshal err:%v key:%s", err, key)
return
}
resp := l.worker.Worker(ctx, taskType, taskId, ed.Data)
l.logger.Infof(ctx, "processTask exec key:%s resp:%+v data:%s", key, resp, str)
option.go
+1 -1
View File
@@ -123,7 +123,7 @@ func WithMaxRetryCount(count int) Option {
func WithMaxWorkers(count int) Option {
return func(o *Options) {
if count < 0 {
count = 10
count = 100
}
o.maxWorkers = count
}
priority/option.go
+8 -2
View File
@@ -18,7 +18,10 @@ type Options struct {
func defaultOptions() Options {
u, _ := uuid.NewV7()
u, err := uuid.NewV7()
if err != nil {
u = uuid.New()
}
return Options{
getInterval: time.Second * 2,
@@ -47,7 +50,10 @@ func WithLogger(log logger.Logger) Option {
// 更新周期
func WithUpdateInterval(d time.Duration) Option {
if d.Abs() < time.Second {
if d <= 0 {
d = time.Second * 5
}
if d < time.Second {
d = time.Second * 5
}
return func(o *Options) {
priority/priority.go
+4 -1
View File
@@ -182,7 +182,10 @@ func (p *Priority) setPriority() (string, error) {
// 解析结果
if resultMap, ok := result.([]interface{}); ok && len(resultMap) == 1 {
resultStr := resultMap[0].(string)
resultStr, ok := resultMap[0].(string)
if !ok {
return "", fmt.Errorf("script unexpected result type: %T", resultMap[0])
}
return resultStr, nil
}
types.go
+1 -1
View File
@@ -41,7 +41,7 @@ type JobData struct {
Minute int // 每小时的第几分钟
Second int // 每分钟的第几秒
CronExpression string // cron表达式
CronSchedule *cron.Schedule // cron表达式解析后的数据
CronSchedule cron.Schedule // cron表达式解析后的数据
}
// 定义各个回调函数