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

优化时间的筛选

Browse Source
This commit is contained in:
Yun
2025-10-04 18:51:22 +08:00
parent 2d6e77352f
commit 737eef2157
7 changed files with 189 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cluster.go
+10 -11
View File
@@ -355,11 +355,11 @@ func (c *Cluster) scheduleTasks() {
// @param extendData 扩展数据
// @return error
func (c *Cluster) EveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
// CreateTime: nowTime,
Day: day,
Hour: hour,
Minute: minute,
@@ -377,11 +377,11 @@ func (c *Cluster) EveryMonth(ctx context.Context, taskId string, day int, hour i
// @param minute int 分钟
// @param second int 秒
func (c *Cluster) EveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryWeek,
CreateTime: nowTime,
// CreateTime: nowTime,
Weekday: week,
Hour: hour,
Minute: minute,
@@ -393,11 +393,11 @@ func (c *Cluster) EveryWeek(ctx context.Context, taskId string, week time.Weekda
// 每天执行一次
func (c *Cluster) EveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryDay,
CreateTime: nowTime,
// CreateTime: nowTime,
Hour: hour,
Minute: minute,
Second: second,
@@ -408,11 +408,11 @@ func (c *Cluster) EveryDay(ctx context.Context, taskId string, hour int, minute
// 每小时执行一次
func (c *Cluster) EveryHour(ctx context.Context, taskId string, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryHour,
CreateTime: nowTime,
// CreateTime: nowTime,
Minute: minute,
Second: second,
}
@@ -422,11 +422,11 @@ func (c *Cluster) EveryHour(ctx context.Context, taskId string, minute int, seco
// 每分钟执行一次
func (c *Cluster) EveryMinute(ctx context.Context, taskId string, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryMinute,
CreateTime: nowTime,
// CreateTime: nowTime,
Second: second,
}
@@ -448,7 +448,6 @@ func (c *Cluster) EverySpace(ctx context.Context, taskId string, spaceTime time.
jobData := JobData{
JobType: JobTypeInterval,
BaseTime: zeroTime, // 默认当天的零点
CreateTime: nowTime,
IntervalTime: spaceTime,
}
error.go
+2
View File
@@ -21,6 +21,8 @@ var (
ErrWeekday = errors.New("weekday must be between Sunday and Saturday")
// 创建时间不能为空
ErrCreateTime = errors.New("create time can not be empty")
// 基准时间不能为空
ErrBaseTime = errors.New("base time can not be empty")
// 间隔时间必须大于0
ErrIntervalTime = errors.New("interval time must be greater than 0")
// 任务Id已存在
next_time.go
+9 -7
View File
@@ -70,8 +70,8 @@ func validateJobData(job JobData) error {
if job.IntervalTime <= 0 {
return ErrIntervalTime
}
if job.CreateTime.IsZero() {
return ErrCreateTime
if job.BaseTime.IsZero() {
return ErrBaseTime
}
}
@@ -88,19 +88,21 @@ func validateJobData(job JobData) error {
return nil
}
// 计算间隔任务下一次执行时间
// 固定基准时间，因为在不同的实例中需要对齐基准点
func calculateNextInterval(t time.Time, job JobData) (*time.Time, error) {
if job.CreateTime.IsZero() {
return nil, ErrCreateTime
if job.BaseTime.IsZero() {
return nil, ErrBaseTime
}
if job.IntervalTime <= 0 {
return nil, ErrIntervalTime
}
// 计算从创建时间到当前时间经过了多少个间隔
elapsed := t.Sub(job.CreateTime)
// 计算从基准时间到当前时间经过了多少个间隔
elapsed := t.Sub(job.BaseTime)
intervals := elapsed / job.IntervalTime
// 计算下一个执行时间
next := job.CreateTime.Add((intervals + 1) * job.IntervalTime)
next := job.BaseTime.Add((intervals + 1) * job.IntervalTime)
// 需要整的
next = next.Round(job.IntervalTime)
next_time_test.go
+39 -22
View File
@@ -10,7 +10,7 @@ import (
func TestGetNextTime(t *testing.T) {
tt := time.Now()
tt := time.Date(2025, 10, 16, 10, 30, 5, 0, time.Local)
// Test cases
tests := []struct {
@@ -40,7 +40,7 @@ func TestGetNextTime(t *testing.T) {
Minute: 0,
Second: 0,
},
expectedTime: time.Date(2025, 9, 23, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022
expectedTime: time.Date(2025, 10, 21, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022
expectedError: nil,
},
{
@@ -51,7 +51,7 @@ func TestGetNextTime(t *testing.T) {
Minute: 0,
Second: 0,
},
expectedTime: time.Date(2025, 9, 18, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022
expectedTime: time.Date(2025, 10, 17, 10, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022
expectedError: nil,
},
{
@@ -61,7 +61,7 @@ func TestGetNextTime(t *testing.T) {
Minute: 0,
Second: 0,
},
expectedTime: time.Date(2025, 9, 17, 16, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
expectedTime: time.Date(2025, 10, 16, 11, 0, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
expectedError: nil,
},
{
@@ -70,17 +70,37 @@ func TestGetNextTime(t *testing.T) {
JobType: JobTypeEveryMinute,
Second: 12,
},
expectedTime: time.Date(tt.Year(), tt.Month(), tt.Day(), tt.Hour(), tt.Minute()+1, 12, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
expectedTime: time.Date(tt.Year(), tt.Month(), tt.Day(), tt.Hour(), tt.Minute(), 12, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
expectedError: nil,
},
{
name: "Test JobTypeInterval",
name: "Test JobTypeIntervalHour",
job: JobData{
JobType: JobTypeInterval,
CreateTime: tt,
BaseTime: tt,
IntervalTime: 1 * time.Hour,
},
expectedTime: tt.Add(1 * time.Hour), // Assuming current date is March 7, 2022, 10:30 AM
expectedTime: time.Date(2025, 10, 16, 12, 00, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
expectedError: nil,
},
{
name: "Test JobTypeIntervalMinute",
job: JobData{
JobType: JobTypeInterval,
BaseTime: tt,
IntervalTime: 1 * time.Minute,
},
expectedTime: time.Date(2025, 10, 16, 10, 31, 0, 0, time.Local), // Assuming current date is March 7, 2022, 10:30 AM
expectedError: nil,
},
{
name: "Test JobTypeIntervalSecond",
job: JobData{
JobType: JobTypeInterval,
BaseTime: tt,
IntervalTime: 1 * time.Second,
},
expectedTime: tt.Add(1 * time.Second), // Assuming current date is March 7, 2022, 10:30 AM
expectedError: nil,
},
{
@@ -95,9 +115,8 @@ func TestGetNextTime(t *testing.T) {
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
now := time.Now()
// loc := time.FixedZone("CST", 8*3600)
nextTime, err := GetNextTime(now, test.job)
nextTime, err := GetNextTime(tt, test.job)
if err != nil {
if test.expectedError == nil || err.Error() != test.expectedError.Error() {
t.Errorf("Expected error: %v, Got error: %v", test.expectedError, err)
@@ -177,7 +196,7 @@ func TestValidateJobData(t *testing.T) {
name: "有效间隔任务",
job: JobData{
JobType: JobTypeInterval,
CreateTime: time.Now(),
BaseTime: time.Now(),
IntervalTime: time.Minute,
Hour: 12,
Minute: 30,
@@ -189,7 +208,7 @@ func TestValidateJobData(t *testing.T) {
name: "无效间隔任务-间隔时间为0",
job: JobData{
JobType: JobTypeInterval,
CreateTime: time.Now(),
BaseTime: time.Now(),
IntervalTime: 0,
Hour: 12,
Minute: 30,
@@ -201,13 +220,13 @@ func TestValidateJobData(t *testing.T) {
name: "无效间隔任务-创建时间为空",
job: JobData{
JobType: JobTypeInterval,
CreateTime: time.Time{},
BaseTime: time.Time{},
IntervalTime: time.Minute,
Hour: 12,
Minute: 30,
Second: 0,
},
expected: ErrCreateTime,
expected: ErrBaseTime,
},
{
name: "无效小时",
@@ -264,7 +283,7 @@ func TestCalculateNextInterval(t *testing.T) {
name: "间隔1小时-当前时间在创建时间之后",
job: JobData{
JobType: JobTypeInterval,
CreateTime: createTime,
BaseTime: createTime,
IntervalTime: time.Hour,
},
currentTime: now,
@@ -274,7 +293,7 @@ func TestCalculateNextInterval(t *testing.T) {
name: "间隔30分钟-刚好在间隔点上",
job: JobData{
JobType: JobTypeInterval,
CreateTime: createTime,
BaseTime: createTime,
IntervalTime: 30 * time.Minute,
},
currentTime: time.Date(2023, 6, 15, 12, 30, 0, 0, time.UTC),
@@ -284,11 +303,11 @@ func TestCalculateNextInterval(t *testing.T) {
name: "间隔1天-跨天",
job: JobData{
JobType: JobTypeInterval,
CreateTime: createTime,
BaseTime: createTime,
IntervalTime: 24 * time.Hour,
},
currentTime: now,
expected: time.Date(2023, 6, 16, 10, 0, 0, 0, time.UTC),
expected: time.Date(2023, 6, 16, 0, 0, 0, 0, time.UTC),
},
}
@@ -301,8 +320,6 @@ func TestCalculateNextInterval(t *testing.T) {
}
}
// 测试月任务
func TestCalculateNextMonthTime(t *testing.T) {
baseTime := time.Date(2023, 6, 15, 12, 30, 45, 0, time.UTC)
@@ -384,7 +401,7 @@ func TestCalculateNextMonthTime(t *testing.T) {
}
}
func TestCalculateNextMonthTimeOnce(t *testing.T){
func TestCalculateNextMonthTimeOnce(t *testing.T) {
// baseTime := time.Date(2023, 6, 15, 12, 30, 45, 0, time.UTC)
tests := []struct {
@@ -673,7 +690,7 @@ func TestGetNextTime_Integration(t *testing.T) {
name: "间隔任务集成测试",
job: JobData{
JobType: JobTypeInterval,
CreateTime: time.Date(2023, 6, 15, 10, 0, 0, 0, time.UTC),
BaseTime: time.Date(2023, 6, 15, 10, 0, 0, 0, time.UTC),
IntervalTime: time.Hour,
},
expected: time.Date(2023, 6, 15, 13, 0, 0, 0, time.UTC),
single.go
+28 -21
View File
@@ -4,7 +4,6 @@ package timerx
import (
"context"
"errors"
"fmt"
"runtime/debug"
"sync"
@@ -67,20 +66,21 @@ func (l *Single) startDaemon() {
}
// 停止所有定时任务
func (s *Single) Stop() {
close(s.stopChan)
func (l *Single) Stop() {
close(l.stopChan)
if s.cancel != nil {
s.cancel()
if l.cancel != nil {
l.cancel()
}
s.wg.Wait()
l.wg.Wait()
l.logger.Infof(l.ctx, "timer single: stopped")
}
// 获取任务数量
func (s *Single) TaskCount() int {
func (l *Single) TaskCount() int {
count := 0
s.workerList.Range(func(k, v interface{}) bool {
l.workerList.Range(func(k, v interface{}) bool {
count++
return true
})
@@ -164,12 +164,12 @@ func (s *Single) cleanupLoop() {
// @return error
func (c *Single) EveryMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryMonth,
TaskId: taskId,
CreateTime: nowTime,
// CreateTime: nowTime,
Day: day,
Hour: hour,
Minute: minute,
@@ -187,12 +187,12 @@ func (c *Single) EveryMonth(ctx context.Context, taskId string, day int, hour in
// @param minute int 分钟
// @param second int 秒
func (c *Single) EveryWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryWeek,
TaskId: taskId,
CreateTime: nowTime,
// CreateTime: nowTime,
Weekday: week,
Hour: hour,
Minute: minute,
@@ -204,12 +204,12 @@ func (c *Single) EveryWeek(ctx context.Context, taskId string, week time.Weekday
// 每天执行一次
func (c *Single) EveryDay(ctx context.Context, taskId string, hour int, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryDay,
TaskId: taskId,
CreateTime: nowTime,
// CreateTime: nowTime,
Hour: hour,
Minute: minute,
Second: second,
@@ -220,12 +220,12 @@ func (c *Single) EveryDay(ctx context.Context, taskId string, hour int, minute i
// 每小时执行一次
func (c *Single) EveryHour(ctx context.Context, taskId string, minute int, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryHour,
TaskId: taskId,
CreateTime: nowTime,
// CreateTime: nowTime,
Minute: minute,
Second: second,
}
@@ -235,12 +235,12 @@ func (c *Single) EveryHour(ctx context.Context, taskId string, minute int, secon
// 每分钟执行一次
func (c *Single) EveryMinute(ctx context.Context, taskId string, second int, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
nowTime := time.Now().In(c.location)
// nowTime := time.Now().In(c.location)
jobData := JobData{
JobType: JobTypeEveryMinute,
TaskId: taskId,
CreateTime: nowTime,
// CreateTime: nowTime,
Second: second,
}
@@ -253,13 +253,17 @@ func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.D
if spaceTime < 0 {
c.logger.Errorf(ctx, "间隔时间不能小于0")
return 0, errors.New("间隔时间不能小于0")
return 0, ErrIntervalTime
}
// 获取当天的零点时间
zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
jobData := JobData{
JobType: JobTypeInterval,
TaskId: taskId,
CreateTime: nowTime,
// CreateTime: nowTime,
BaseTime: zeroTime,
IntervalTime: spaceTime,
}
@@ -417,10 +421,13 @@ func (s *Single) executeTask(ctx context.Context, timer timerStr, originTime tim
}()
// 执行回调
begin := time.Now()
if err := s.doTask(traceCtx, timer, originTime); err != nil {
s.logger.Errorf(traceCtx, "timer: 任务执行失败: %s", err.Error())
}
s.logger.Infof(traceCtx, "timer Single end taskId:%s originTime:%d cost:%dms", timer.TaskId, originTime.UnixMilli(), time.Since(begin).Milliseconds())
case <-traceCtx.Done():
s.logger.Errorf(traceCtx, "timer: 任务执行超时: %s", timer.TaskId)
default:
// 任务正在执行中，跳过本次
s.logger.Infof(traceCtx, "timer: 任务正在执行中，跳过本次 %s", timer.TaskId)
single_test.go
+56 -16
View File
@@ -3,6 +3,7 @@ package timerx_test
import (
"context"
"fmt"
"strings"
"sync"
"sync/atomic"
"testing"
@@ -123,18 +124,18 @@ func TestSingleTimer_Deduplication(t *testing.T) {
assert.NoError(t, err)
// 等待一段时间，检查去重是否生效
time.Sleep(200 * time.Millisecond)
time.Sleep(250 * time.Millisecond)
// 应该只有1次执行（因为任务执行需要100ms，50ms的间隔会被去重）
assert.Equal(t, int32(1), atomic.LoadInt32(&executionCount))
t.Logf("warn: %v", mockLogger.Warns)
t.Logf("info: %v", mockLogger.Infos)
// t.Logf("warn: %+v", mockLogger.Warns)
// t.Logf("info: %+v", mockLogger.Infos)
fmt.Println("info:", mockLogger.Infos)
fmt.Println("warn:", mockLogger.Warns)
// 检查是否有去重日志
assert.Contains(t, mockLogger.Warns, "timer: 任务已执行，跳过本次执行 dedup-test")
assert.Contains(t, mockLogger.Infos, "timer: 任务正在执行中，跳过本次 dedup-test")
}
// 测试并发安全
@@ -190,11 +191,12 @@ func TestSingleTimer_Timeout(t *testing.T) {
ctx := context.Background()
mockLogger := &MockLogger{}
timer := timerx.InitSingle(ctx, timerx.WithLogger(mockLogger))
timer := timerx.InitSingle(ctx, timerx.WithLogger(mockLogger), timerx.WithTimeout(1*time.Second))
defer timer.Stop()
// 长时间运行的任务
longTask := func(ctx context.Context, data interface{}) error {
fmt.Println("long task start")
select {
case <-time.After(2 * time.Second): // 超过超时时间
case <-ctx.Done():
@@ -206,10 +208,20 @@ func TestSingleTimer_Timeout(t *testing.T) {
_, err := timer.EverySpace(ctx, "timeout-test", 100*time.Millisecond, longTask, nil)
assert.NoError(t, err)
time.Sleep(500 * time.Millisecond)
time.Sleep(time.Second * 5)
// 检查是否有超时相关的错误日志
assert.Contains(t, mockLogger.Errors, "context deadline exceeded")
if len(mockLogger.Errors) == 0 {
t.Fatalf("expected timeout error log, got none")
}
isTimeout := false
for _, err := range mockLogger.Errors {
isTimeout = strings.Contains(err, "context deadline exceeded")
if isTimeout {
break
}
}
assert.True(t, isTimeout)
}
// 测试panic恢复
@@ -230,7 +242,18 @@ func TestSingleTimer_PanicRecovery(t *testing.T) {
time.Sleep(200 * time.Millisecond)
// 检查是否有panic恢复日志
assert.Contains(t, mockLogger.Errors, "timer: 回调任务panic err")
if len(mockLogger.Errors) == 0 {
t.Fatalf("expected panic recovery log, got none")
}
isPanic := false
for _, err := range mockLogger.Errors {
isPanic = strings.Contains(err, "timer Single call panic err")
if isPanic {
break
}
}
assert.True(t, isPanic)
}
// 测试不同时间类型的任务
@@ -274,12 +297,12 @@ func TestSingleTimer_DifferentJobTypes(t *testing.T) {
}, nil)
assert.NoError(t, err)
time.Sleep(300 * time.Millisecond)
time.Sleep(time.Second)
// 只有间隔任务应该执行
assert.Equal(t, int32(1), atomic.LoadInt32(&counts.space))
assert.Equal(t, int32(0), atomic.LoadInt32(&counts.month))
assert.Equal(t, int32(0), atomic.LoadInt32(&counts.week))
assert.Equal(t, int32(9), atomic.LoadInt32(&counts.space))
assert.Equal(t, int32(1), atomic.LoadInt32(&counts.month))
assert.Equal(t, int32(1), atomic.LoadInt32(&counts.week))
}
// 测试上下文取消
@@ -329,6 +352,7 @@ func TestSingleTimer_ExtendData(t *testing.T) {
_, err := timer.EverySpace(ctx, "data-test", 100*time.Millisecond,
func(ctx context.Context, data interface{}) error {
fmt.Println("data:", data)
if data != nil {
receivedData = data.(*TestData)
}
@@ -336,7 +360,9 @@ func TestSingleTimer_ExtendData(t *testing.T) {
}, testData)
assert.NoError(t, err)
time.Sleep(150 * time.Millisecond)
time.Sleep(time.Second)
t.Logf("receivedData: %+v", receivedData)
assert.NotNil(t, receivedData)
assert.Equal(t, "hello", receivedData.Message)
@@ -392,11 +418,15 @@ func TestGetNextTime2(t *testing.T) {
jobData := timerx.JobData{
JobType: timerx.JobTypeInterval,
IntervalTime: time.Minute,
// CreateTime: now,
BaseTime: now,
}
tt := time.Date(now.Year(), now.Month(), now.Day(), now.Hour(), now.Minute(), 0, 0, time.UTC)
nextTime, err := timerx.GetNextTime(now, jobData)
assert.NoError(t, err)
assert.WithinDuration(t, now.Add(time.Minute), *nextTime, time.Second)
assert.WithinDuration(t, tt.Add(time.Minute), *nextTime, time.Second)
}
// 基准测试
@@ -433,7 +463,16 @@ func TestSingleTimer_Logging(t *testing.T) {
// 检查日志记录
assert.NotEmpty(t, mockLogger.Errors)
assert.Contains(t, mockLogger.Errors[0], "timer: 回调任务panic err")
if len(mockLogger.Errors) == 0 {
t.Fatalf("expected panic recovery log, got none")
}
isPanic := false
for _, err := range mockLogger.Errors {
isPanic = strings.Contains(err, "test panic for logging")
}
assert.True(t, isPanic)
}
// 测试时区处理
@@ -457,12 +496,13 @@ func TestSingleTimer_Timezone(t *testing.T) {
// 添加下一秒执行的任务
_, err := timer.EverySpace(ctx, "tz-test", time.Second,
func(ctx context.Context, data interface{}) error {
fmt.Println("executed in location:", loc)
executed = true
return nil
}, nil)
assert.NoError(t, err)
time.Sleep(1500 * time.Millisecond)
time.Sleep(5 * time.Second)
assert.True(t, executed)
})
}
types.go
-1
View File
@@ -30,7 +30,6 @@ type JobData struct {
TaskId string // 任务ID 全局唯一键(only cluster)
NextTime time.Time // 下次执行时间
BaseTime time.Time // 基准时间(间隔的基准时间)
CreateTime time.Time // 任务创建时间
IntervalTime time.Duration // 任务间隔时间
Month time.Month // 每年的第几个月
Weekday time.Weekday // 每周的周几