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

大版本更新

Browse Source
This commit is contained in:
Yun
2024-05-20 09:35:12 +08:00
parent 0921514339
commit 7912bbc56c
7 changed files with 513 additions and 206 deletions
Download Patch File Download Diff File Expand all files Collapse all files
single.go
+151 -59
View File
@@ -5,7 +5,6 @@ package timerx
import (
"context"
"errors"
"fmt"
"runtime/debug"
"sync"
"time"
@@ -15,12 +14,12 @@ import (
// 1. 这个定时器的作用范围是本机
// 2. 适用简单的时间间隔定时任务
// uuid -> timerStr
var timerMap = make(map[string]*timerStr)
var timerMapMux sync.Mutex
// 定时器结构体
var singleWorkerList sync.Map
var timerCount int // 当前定时数目
var onceLimit sync.Once // 实现单例
var singleTimerIndex int // 当前定时数目
var singleOnceLimit sync.Once // 实现单例
type Single struct {
ctx context.Context
@@ -29,10 +28,13 @@ type Single struct {
var sin *Single = nil
// 定时器类
func InitSingle(ctx context.Context, opts ...Option) *Single {
onceLimit.Do(func() {
var singleNextTime = time.Now() // 下一次执行的时间
// 定时器类
// @param ctx context.Context 上下文
// @param opts ...Option 配置项
func InitSingle(ctx context.Context, opts ...Option) *Single {
singleOnceLimit.Do(func() {
op := newOptions(opts...)
sin = &Single{
@@ -46,7 +48,7 @@ func InitSingle(ctx context.Context, opts ...Option) *Single {
for {
select {
case t := <-timer.C:
if t.Before(nextTime) {
if t.Before(singleNextTime) {
// 当前时间小于下次发送时间：跳过
continue
}
@@ -65,89 +67,175 @@ func InitSingle(ctx context.Context, opts ...Option) *Single {
return sin
}
// 每月执行一次
// @param ctx 上下文
// @param taskId 任务ID
// @param day 每月的几号
// @param hour 小时
// @param minute 分钟
// @param second 秒
// @param callback 回调函数
// @param extendData 扩展数据
// @return error
func (c *Single) AddMonth(ctx context.Context, taskId string, day int, hour int, minute int, second int, callback callback, extendData interface{}) (int, error) {
nowTime := time.Now()
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
Day: day,
Hour: hour,
Minute: minute,
Second: second,
}
return c.addJob(ctx, jobData, callback, extendData)
}
// 每周执行一次
// @param ctx context.Context 上下文
// @param taskId string 任务ID
// @param week time.Weekday 周
// @param hour int 小时
// @param minute int 分钟
// @param second int 秒
func (c *Single) AddWeek(ctx context.Context, taskId string, week time.Weekday, hour int, minute int, second int, callback callback, extendData interface{}) (int, error) {
nowTime := time.Now()
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
Weekday: week,
Hour: hour,
Minute: minute,
Second: second,
}
return c.addJob(ctx, jobData, callback, extendData)
}
// 每天执行一次
func (c *Single) AddDay(ctx context.Context, taskId string, hour int, minute int, second int, callback callback, extendData interface{}) (int, error) {
nowTime := time.Now()
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
Hour: hour,
Minute: minute,
Second: second,
}
return c.addJob(ctx, jobData, callback, extendData)
}
// 每小时执行一次
func (c *Single) AddHour(ctx context.Context, taskId string, minute int, second int, callback callback, extendData interface{}) (int, error) {
nowTime := time.Now()
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
Minute: minute,
Second: second,
}
return c.addJob(ctx, jobData, callback, extendData)
}
// 每分钟执行一次
func (c *Single) AddMinute(ctx context.Context, taskId string, second int, callback callback, extendData interface{}) (int, error) {
nowTime := time.Now()
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
Second: second,
}
return c.addJob(ctx, jobData, callback, extendData)
}
// 特定时间间隔
func (c *Single) AddSpace(ctx context.Context, taskId string, spaceTime time.Duration, callback callback, extendData interface{}) (int, error) {
nowTime := time.Now()
if spaceTime < 0 {
c.logger.Errorf(ctx, "间隔时间不能小于0")
return 0, errors.New("间隔时间不能小于0")
}
jobData := JobData{
JobType: JobTypeEveryMonth,
CreateTime: nowTime,
IntervalTime: spaceTime,
}
return c.addJob(ctx, jobData, callback, extendData)
}
// 间隔定时器
// @param space 间隔时间
// @param call 回调函数
// @param extend 附加参数
// @return int 定时器索引
// @return error 错误
func (s *Single) Add(space time.Duration, call callback, extend interface{}) (int, error) {
timerMapMux.Lock()
defer timerMapMux.Unlock()
if space != space.Abs() {
s.logger.Errorf(s.ctx, "space must be positive")
return 0, errors.New("space must be positive")
}
timerCount += 1
func (l *Single) addJob(ctx context.Context, jobData JobData, call callback, extend interface{}) (int, error) {
singleTimerIndex += 1
nowTime := time.Now()
_, err := GetNextTime(nowTime, time.Local, jobData)
if err != nil {
l.logger.Errorf(ctx, "获取下次执行时间失败:%s", err.Error())
return 0, err
}
t := timerStr{
Callback: call,
BeginTime: nowTime,
NextTime: nowTime, // nowTime.Add(space), // 添加任务的时候就执行一次
SpaceTime: space,
CanRunning: make(chan struct{}, 1),
ExtendData: extend,
JobData: &jobData,
}
timerMap[fmt.Sprintf("%d", timerCount)] = &t
singleWorkerList.Store(singleTimerIndex, t)
if t.NextTime.Before(nextTime) {
// 本条规则下次需要发送的时间小于系统下次发送时间：替换
nextTime = t.NextTime
}
return timerCount, nil
return singleTimerIndex, nil
}
// 删除定时器
func (s *Single) Del(index string) {
timerMapMux.Lock()
defer timerMapMux.Unlock()
delete(timerMap, index)
func (s *Single) Del(index int) {
singleWorkerList.Delete(index)
}
// 迭代定时器列表
func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
timerMapMux.Lock()
defer timerMapMux.Unlock()
// fmt.Println("nowTime:", nowTime.Format("2006-01-02 15:04:05.000"))
// 默认5秒后（如果没有值就暂停进来5秒）
newNextTime := nowTime.Add(time.Second * 5)
index := 0
for _, v := range timerMap {
singleWorkerList.Range(func(k, v interface{}) bool {
index++
v := v
// 判断执行的时机
if v.NextTime.Before(nowTime) {
// fmt.Println("NextTime", v.NextTime.Format("2006-01-02 15:04:05.000"))
timeStr := v.(timerStr)
v.NextTime = v.NextTime.Add(v.SpaceTime)
// 判断下次执行时间与当前时间
if v.NextTime.Before(nowTime) {
v.NextTime = nowTime.Add(v.SpaceTime)
}
if timeStr.JobData.NextTime.Before(nowTime) || timeStr.JobData.NextTime.Equal(nowTime) {
// 可执行
nextTime, _ := GetNextTime(nowTime, time.Local, *timeStr.JobData)
timeStr.JobData.NextTime = *nextTime
if index == 1 {
// 循环的第一个需要替换默认值
newNextTime = v.NextTime
newNextTime = timeStr.JobData.NextTime
}
// 获取最小的
if v.NextTime.Before(newNextTime) {
if nextTime.Before(newNextTime) {
// 本规则下次发送时间小于系统下次需要执行的时间：替换
newNextTime = v.NextTime
newNextTime = *nextTime
}
// 处理中就跳过本次
go func(ctx context.Context, v *timerStr) {
go func(ctx context.Context, v timerStr) {
select {
case v.CanRunning <- struct{}{}:
defer func() {
@@ -165,18 +253,22 @@ func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
// fmt.Printf("timer: 已在执行 %v %v \n", k, v.Tag)
return
}
}(ctx, v)
}(ctx, timeStr)
}
}
return true
})
// 实际下次时间小于预期下次时间：替换
if nextTime.Before(newNextTime) {
if singleNextTime.Before(newNextTime) {
// 判断一下避免异常
if newNextTime.Before(nowTime) {
// 比当前时间小
nextTime = nowTime
singleNextTime = nowTime
} else {
nextTime = newNextTime
singleNextTime = newNextTime
}
}