Compare commits
23 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 yun
|
362d1f455a | ||
|
yun
|
fbb74cdd6d | ||
|
yun
|
319d6b6db1 | ||
 Administrator
|
3cc3f0400b | ||
|
Administrator
|
5793afbab7 | ||
|
Administrator
|
30093d0717 | ||
|
Administrator
|
a4a0d86d74 | ||
|
Administrator
|
80bd6b4327 | ||
|
Administrator
|
b9444c8bb6 | ||
|
Administrator
|
0dbd1ee9c2 | ||
|
Administrator
|
aabce29211 | ||
|
Administrator
|
7475f9cd3b | ||
|
Administrator
|
fabf7f65d7 | ||
|
Administrator
|
e6915aa766 | ||
|
Administrator
|
de4a9c8f31 | ||
|
Administrator
|
356f843747 | ||
|
Administrator
|
52ed316cd1 | ||
|
Administrator
|
778bf75650 | ||
|
Administrator
|
a87fee1f38 | ||
|
Administrator
|
1a70738e6d | ||
|
yun
|
c929d1a57d | ||
|
yun
|
4944efbf29 | ||
|
yun
|
e57f941001 |
+24
-25
@@ -1,26 +1,25 @@
|
||||
# You can override the included template(s) by including variable overrides
|
||||
# SAST customization: https://docs.gitlab.com/ee/user/application_security/sast/#customizing-the-sast-settings
|
||||
# Secret Detection customization: https://docs.gitlab.com/ee/user/application_security/secret_detection/#customizing-settings
|
||||
# Dependency Scanning customization: https://docs.gitlab.com/ee/user/application_security/dependency_scanning/#customizing-the-dependency-scanning-settings
|
||||
# Container Scanning customization: https://docs.gitlab.com/ee/user/application_security/container_scanning/#customizing-the-container-scanning-settings
|
||||
# Note that environment variables can be set in several places
|
||||
# See https://docs.gitlab.com/ee/ci/variables/#cicd-variable-precedence
|
||||
stages:
|
||||
- build
|
||||
- test
|
||||
- deploy
|
||||
- review
|
||||
- dast
|
||||
- staging
|
||||
- canary
|
||||
- production
|
||||
- incremental rollout 10%
|
||||
- incremental rollout 25%
|
||||
- incremental rollout 50%
|
||||
- incremental rollout 100%
|
||||
- performance
|
||||
- cleanup
|
||||
sast:
|
||||
stage: test
|
||||
include:
|
||||
- template: Auto-DevOps.gitlab-ci.yml
|
||||
- deploy
|
||||
|
||||
before_script:
|
||||
- 'which ssh-agent || ( apt-get update -y && apt-get install openssh-client -y )'
|
||||
- eval $(ssh-agent -s)
|
||||
- echo "$SSH_PRIVATE_KEY" | tr -d '\r' | ssh-add -
|
||||
- mkdir -p ~/.ssh
|
||||
- chmod 700 ~/.ssh
|
||||
- '[[ -f /.dockerenv ]] && echo -e "Host *\n\tStrictHostKeyChecking no\n\n" > ~/.ssh/config'
|
||||
|
||||
deploy_job:
|
||||
stage: deploy
|
||||
script:
|
||||
- git remote remove github || true
|
||||
- git remote add github git@github.com:yun-ink/timerx.git
|
||||
- git remote -v
|
||||
- git checkout master
|
||||
- git fsck --full
|
||||
- git prune
|
||||
- git gc --prune=now --aggressive
|
||||
- git push -u github master -f
|
||||
only:
|
||||
- master
|
||||
- tags
|
||||
+24
-45
@@ -1,4 +1,4 @@
|
||||
package timer
|
||||
package timerx
|
||||
|
||||
import (
|
||||
"context"
|
||||
@@ -10,17 +10,23 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"code.yun.ink/open/timer/lockx"
|
||||
"code.yun.ink/pkg/lockx"
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
// 功能描述
|
||||
// 这是基于Redis的定时任务调度器，能够有效的在服务集群里面调度任务，避免了单点压力过高或单点故障问题
|
||||
// 由于所有的服务代码是一致的，也就是一个定时任务将在所有的服务都有注册，具体调度到哪个服务运行看调度结果
|
||||
|
||||
// 暂不支持删除定时器，因为这个定时器的设计是基于全局的，如果删除了，那么其他服务就不知道了
|
||||
|
||||
// 单例模式
|
||||
var clusterOnceLimit sync.Once
|
||||
|
||||
// 已注册的任务列表
|
||||
var clusterWorkerList sync.Map
|
||||
|
||||
type cluster struct {
|
||||
type Cluster struct {
|
||||
ctx context.Context
|
||||
redis *redis.Client
|
||||
lockKey string // 全局计算的key
|
||||
@@ -29,14 +35,14 @@ type cluster struct {
|
||||
listKey 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) *Cluster {
|
||||
clusterOnceLimit.Do(func() {
|
||||
clu = &cluster{
|
||||
clu = &Cluster{
|
||||
ctx: ctx,
|
||||
redis: red,
|
||||
lockKey: "timer:cluster_globalLockKey",
|
||||
lockKey: "timer:cluster_globalLockKey", // 定时器的全局锁
|
||||
nextKey: "timer:cluster_nextKey",
|
||||
zsetKey: "timer:cluster_zsetKey",
|
||||
listKey: "timer:cluster_listKey",
|
||||
@@ -63,7 +69,7 @@ func InitCluster(ctx context.Context, red *redis.Client) *cluster {
|
||||
return clu
|
||||
}
|
||||
|
||||
func (c *cluster) AddTimer(ctx context.Context, uniqueKey string, spaceTime time.Duration, callback callback, extend ExtendParams) error {
|
||||
func (c *Cluster) Add(ctx context.Context, uniqueKey string, spaceTime time.Duration, callback callback, extendData interface{}) error {
|
||||
_, ok := clusterWorkerList.Load(uniqueKey)
|
||||
if ok {
|
||||
return errors.New("key已存在")
|
||||
@@ -86,12 +92,12 @@ func (c *cluster) AddTimer(ctx context.Context, uniqueKey string, spaceTime time
|
||||
nowTime := time.Now()
|
||||
|
||||
t := timerStr{
|
||||
BeginTime: nowTime,
|
||||
NextTime: nowTime,
|
||||
SpaceTime: spaceTime,
|
||||
Callback: callback,
|
||||
Extend: extend,
|
||||
UniqueKey: uniqueKey,
|
||||
BeginTime: nowTime,
|
||||
NextTime: nowTime,
|
||||
SpaceTime: spaceTime,
|
||||
Callback: callback,
|
||||
ExtendData: extendData,
|
||||
UniqueKey: uniqueKey,
|
||||
}
|
||||
|
||||
clusterWorkerList.Store(uniqueKey, t)
|
||||
@@ -119,7 +125,7 @@ func (c *cluster) AddTimer(ctx context.Context, uniqueKey string, spaceTime time
|
||||
}
|
||||
|
||||
// 计算下一次执行的时间
|
||||
func (c *cluster) getNextTime() {
|
||||
func (c *Cluster) getNextTime() {
|
||||
|
||||
// log.Println("begin computer")
|
||||
ctx, cancel := context.WithCancel(c.ctx)
|
||||
@@ -184,31 +190,8 @@ func getNextExecTime(beforeTime time.Time, spaceTime time.Duration) time.Time {
|
||||
}
|
||||
|
||||
// 获取任务
|
||||
func (c *cluster) getTask() {
|
||||
func (c *Cluster) getTask() {
|
||||
// 定时去Redis获取任务
|
||||
// zb := redis.ZRangeBy{
|
||||
// Min: "0",
|
||||
// Max: fmt.Sprintf("%+v", time.Now().UnixMilli()),
|
||||
// }
|
||||
|
||||
// taskList, _ := c.redis.ZRangeByScore(c.ctx, c.zsetKey, &zb).Result()
|
||||
|
||||
// if len(taskList) == 0 {
|
||||
// return
|
||||
// }
|
||||
|
||||
// p := c.redis.Pipeline()
|
||||
|
||||
// for _, val := range taskList {
|
||||
// // 添加到可执行队列
|
||||
// p.LPush(c.ctx, c.listKey, val)
|
||||
// // 删除有序集合
|
||||
// p.ZRem(c.ctx, c.zsetKey, val)
|
||||
// }
|
||||
// _, err := p.Exec(c.ctx)
|
||||
// // fmt.Println(err)
|
||||
// _ = err
|
||||
|
||||
script := `
|
||||
local token = redis.call('zrangebyscore',KEYS[1],ARGV[1],ARGV[2])
|
||||
for i,v in ipairs(token) do
|
||||
@@ -222,7 +205,7 @@ func (c *cluster) getTask() {
|
||||
}
|
||||
|
||||
// 监听任务
|
||||
func (c *cluster) watch() {
|
||||
func (c *Cluster) watch() {
|
||||
// 执行任务
|
||||
for {
|
||||
keys, err := c.redis.BLPop(c.ctx, time.Second*10, c.listKey).Result()
|
||||
@@ -236,8 +219,6 @@ func (c *cluster) watch() {
|
||||
|
||||
// 执行任务
|
||||
func doTask(ctx context.Context, red *redis.Client, taskId string) {
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
@@ -262,8 +243,6 @@ func doTask(ctx context.Context, red *redis.Client, taskId string) {
|
||||
}
|
||||
defer lock.Unlock()
|
||||
|
||||
ctx = context.WithValue(ctx, extendParamKey, t.Extend)
|
||||
|
||||
// 执行任务
|
||||
t.Callback(ctx)
|
||||
t.Callback(ctx,t.ExtendData)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package timer
|
||||
package timerx
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
+15
-56
@@ -5,7 +5,7 @@ import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"code.yun.ink/open/timer"
|
||||
"code.yun.ink/pkg/timerx"
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
@@ -30,7 +30,7 @@ func main() {
|
||||
|
||||
func worker() {
|
||||
client := getRedis()
|
||||
w := timer.InitWorker(context.Background(), client, &Worker{})
|
||||
w := timerx.InitOnce(context.Background(), client, &Worker{})
|
||||
w.Add("test", "test", 1*time.Second, map[string]interface{}{
|
||||
"test": "test",
|
||||
})
|
||||
@@ -52,11 +52,11 @@ func worker() {
|
||||
|
||||
type Worker struct{}
|
||||
|
||||
func (w *Worker) Worker(uniqueKey string, jobType string,data map[string]interface{}) timer.WorkerCode {
|
||||
func (w *Worker) Worker(uniqueKey string, jobType 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)
|
||||
return timer.WorkerCodeAgain
|
||||
return timerx.WorkerCodeAgain,time.Second
|
||||
}
|
||||
|
||||
func getRedis() *redis.Client {
|
||||
@@ -76,63 +76,22 @@ func re() {
|
||||
client := getRedis()
|
||||
|
||||
ctx := context.Background()
|
||||
cl := timer.InitCluster(ctx, client)
|
||||
cl.AddTimer(ctx, "test1", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text1",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test2", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text2",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test3", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text3",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test4", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text4",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test5", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text5",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test6", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text6",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test7", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text7",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test8", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text8",
|
||||
},
|
||||
})
|
||||
cl.AddTimer(ctx, "test9", 1*time.Millisecond, aa, timer.ExtendParams{
|
||||
Params: map[string]interface{}{
|
||||
"test": "text9",
|
||||
},
|
||||
})
|
||||
cl := timerx.InitCluster(ctx, client)
|
||||
cl.Add(ctx, "test1", 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, "test4", 1*time.Millisecond, aa, "data")
|
||||
cl.Add(ctx, "test5", 1*time.Millisecond, aa, "data")
|
||||
cl.Add(ctx, "test6", 1*time.Millisecond, aa, "data")
|
||||
|
||||
select {}
|
||||
}
|
||||
|
||||
func aa(ctx context.Context) bool {
|
||||
// fmt.Println(time.Now().Format(time.RFC3339))
|
||||
// fmt.Println("gggggggggggggggggggggggggggg")
|
||||
a, err := timer.GetExtendParams(ctx)
|
||||
fmt.Printf("%+v %+v \n\n", a, err)
|
||||
func aa(ctx context.Context, data interface{}) error {
|
||||
fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
|
||||
fmt.Println(data)
|
||||
time.Sleep(time.Second * 5)
|
||||
return true
|
||||
return nil
|
||||
}
|
||||
|
||||
func d() {
|
||||
|
||||
@@ -1,13 +1,11 @@
|
||||
module code.yun.ink/open/timer
|
||||
module code.yun.ink/pkg/timerx
|
||||
|
||||
go 1.19
|
||||
|
||||
require (
|
||||
github.com/go-redis/redis/v8 v8.11.5
|
||||
github.com/gomodule/redigo v1.8.9
|
||||
)
|
||||
require github.com/go-redis/redis/v8 v8.11.5
|
||||
|
||||
require (
|
||||
code.yun.ink/pkg/lockx v1.0.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
|
||||
)
|
||||
|
||||
@@ -1,27 +1,17 @@
|
||||
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=
|
||||
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
|
||||
github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
|
||||
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
|
||||
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
|
||||
github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
|
||||
github.com/go-redis/redis/v8 v8.11.5 h1:AcZZR7igkdvfVmQTPnu9WE37LRrO/YrBH5zWyjDC0oI=
|
||||
github.com/go-redis/redis/v8 v8.11.5/go.mod h1:gREzHqY1hg6oD9ngVRbLStwAWKhA0FEgq8Jd4h5lpwo=
|
||||
github.com/gomodule/redigo v1.8.9 h1:Sl3u+2BI/kk+VEatbj0scLdrFhjPmbxOc1myhDP41ws=
|
||||
github.com/gomodule/redigo v1.8.9/go.mod h1:7ArFNvsTjH8GMMzB4uy1snslv2BwmginuMs06a1uzZE=
|
||||
github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
|
||||
github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
|
||||
github.com/onsi/gomega v1.18.1 h1:M1GfJqGRrBrrGGsbxzV5dqM2U2ApXefZCQpkukxYRLE=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5CcY=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0=
|
||||
golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e h1:fLOSk5Q00efkSvAm+4xcoXD+RRmLmmulPn5I3Y9F2EM=
|
||||
golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
|
||||
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ=
|
||||
gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c h1:dUUwHk2QECo/6vqA44rthZ8ie2QXMNeKRTHCNY2nXvo=
|
||||
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
|
||||
|
||||
-120
@@ -1,120 +0,0 @@
|
||||
package lockx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log"
|
||||
"time"
|
||||
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
// 全局锁
|
||||
type globalLock struct {
|
||||
redis *redis.Client
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
uniqueKey string
|
||||
value string
|
||||
}
|
||||
|
||||
func NewGlobalLock(ctx context.Context, red *redis.Client, uniqueKey string) *globalLock {
|
||||
ctx, cancel := context.WithTimeout(ctx, time.Second*30)
|
||||
return &globalLock{
|
||||
redis: red,
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
uniqueKey: uniqueKey,
|
||||
value: fmt.Sprintf("%d", time.Now().UnixNano()),
|
||||
}
|
||||
}
|
||||
|
||||
// 获取锁
|
||||
func (g *globalLock) Lock() bool {
|
||||
|
||||
script := `
|
||||
local token = redis.call('get',KEYS[1])
|
||||
if token == false
|
||||
then
|
||||
return redis.call('set',KEYS[1],ARGV[1],'EX',ARGV[2])
|
||||
end
|
||||
return 'ERROR'
|
||||
`
|
||||
|
||||
resp, err := g.redis.Eval(g.ctx, script, []string{g.uniqueKey}, g.value, 5).Result()
|
||||
if resp != "OK" {
|
||||
_ = err
|
||||
log.Println("globalLock Lock", resp, err, g.uniqueKey, g.value)
|
||||
}
|
||||
if resp == "OK" {
|
||||
g.refresh()
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// 尝试获取锁
|
||||
func (g *globalLock) Try(limitTimes int) bool {
|
||||
for i := 0; i < limitTimes; i++ {
|
||||
if g.Lock() {
|
||||
return true
|
||||
}
|
||||
time.Sleep(time.Millisecond * 100)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// 删除锁
|
||||
func (g *globalLock) Unlock() bool {
|
||||
|
||||
script := `
|
||||
local token = redis.call('get',KEYS[1])
|
||||
if token == ARGV[1]
|
||||
then
|
||||
redis.call('del',KEYS[1])
|
||||
return 'OK'
|
||||
end
|
||||
return 'ERROR'
|
||||
`
|
||||
|
||||
resp, err := g.redis.Eval(g.ctx, script, []string{g.uniqueKey}, g.value).Result()
|
||||
if resp != "OK" {
|
||||
log.Println("globalLock Unlock", resp, err, g.uniqueKey, g.value)
|
||||
}
|
||||
g.cancel()
|
||||
return false
|
||||
}
|
||||
|
||||
// 刷新锁
|
||||
func (g *globalLock) refresh() {
|
||||
go func() {
|
||||
t := time.NewTicker(time.Second)
|
||||
for {
|
||||
select {
|
||||
case <-t.C:
|
||||
g.refreshExec()
|
||||
case <-g.ctx.Done():
|
||||
t.Stop()
|
||||
return
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (g *globalLock) refreshExec() bool {
|
||||
script := `
|
||||
local token = redis.call('get',KEYS[1])
|
||||
if token == ARGV[1]
|
||||
then
|
||||
redis.call('set',KEYS[1],ARGV[1],'EX',ARGV[2])
|
||||
return 'OK'
|
||||
end
|
||||
return 'ERROR'
|
||||
`
|
||||
|
||||
resp, err := g.redis.Eval(g.ctx, script, []string{g.uniqueKey}, g.value, 5).Result()
|
||||
if resp != "OK" {
|
||||
log.Println("globalLock refresh", resp, err, g.uniqueKey, g.value)
|
||||
}
|
||||
return resp == "OK"
|
||||
}
|
||||
@@ -1,52 +0,0 @@
|
||||
package lockx_test
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"code.yun.ink/open/timer/lockx"
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
var Redis *redis.Client
|
||||
|
||||
// func TestMain(m *testing.M) {
|
||||
// client := redis.NewClient(&redis.Options{
|
||||
// Addr: "127.0.0.1" + ":" + "6379",
|
||||
// Password: "", // no password set
|
||||
// DB: 0, // use default DB
|
||||
// })
|
||||
// if client == nil {
|
||||
// fmt.Println("redis init error")
|
||||
// return
|
||||
// }
|
||||
// // fmt.Println("ffff")
|
||||
// Redis = client
|
||||
// }
|
||||
|
||||
func TestLockx(t *testing.T) {
|
||||
client := redis.NewClient(&redis.Options{
|
||||
Addr: "127.0.0.1" + ":" + "6379",
|
||||
Password: "", // no password set
|
||||
DB: 0, // use default DB
|
||||
})
|
||||
if client == nil {
|
||||
fmt.Println("redis init error")
|
||||
return
|
||||
}
|
||||
fmt.Println("begin")
|
||||
ctx := context.Background()
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
lock := lockx.NewGlobalLock(ctx, client, "lockx:test")
|
||||
|
||||
if !lock.Lock() {
|
||||
fmt.Println("lock error")
|
||||
}
|
||||
defer lock.Unlock()
|
||||
|
||||
fmt.Println("ssss")
|
||||
|
||||
}
|
||||
+59
-34
@@ -1,9 +1,11 @@
|
||||
package timer
|
||||
package timerx
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -11,25 +13,34 @@ import (
|
||||
"github.com/go-redis/redis/v8"
|
||||
)
|
||||
|
||||
// 功能描述
|
||||
// 1. 任务全局唯一
|
||||
// 2. 任务只执行一次
|
||||
// 3. 任务执行失败可以重新放入队列
|
||||
|
||||
// 单次的任务队列
|
||||
type worker struct {
|
||||
ctx context.Context
|
||||
zsetKey string
|
||||
listKey string
|
||||
redis *redis.Client
|
||||
worker WorkerInterface
|
||||
worker Callback
|
||||
}
|
||||
|
||||
type WorkerCode int
|
||||
|
||||
const (
|
||||
WorkerCodeSuccess WorkerCode = 0
|
||||
WorkerCodeAgain WorkerCode = -1
|
||||
WorkerCodeSuccess WorkerCode = 0 // 处理完成(不需要重入)
|
||||
WorkerCodeAgain WorkerCode = -1 // 需要继续定时,默认原来的时间
|
||||
)
|
||||
|
||||
// 需要考虑执行失败重新放入队列的情况
|
||||
type WorkerInterface interface {
|
||||
Worker(uniqueKey string, jobType string, data map[string]interface{}) WorkerCode
|
||||
type Callback interface {
|
||||
// 任务执行
|
||||
// uniqueKey: 任务唯一标识
|
||||
// jobType: 任务类型，用于区分任务
|
||||
// data: 任务数据
|
||||
Worker(uniqueKey string, jobType string, data interface{}) (WorkerCode, time.Duration)
|
||||
}
|
||||
|
||||
var wo *worker = nil
|
||||
@@ -37,21 +48,22 @@ var once sync.Once
|
||||
|
||||
type extendData struct {
|
||||
Delay time.Duration
|
||||
Data map[string]interface{}
|
||||
Data interface{}
|
||||
}
|
||||
|
||||
func InitWorker(ctx context.Context, re *redis.Client, w WorkerInterface) *worker {
|
||||
// 初始化
|
||||
func InitOnce(ctx context.Context, re *redis.Client, w Callback) *worker {
|
||||
|
||||
once.Do(func() {
|
||||
wo = &worker{
|
||||
ctx: ctx,
|
||||
zsetKey: "timer:job_zsetkey",
|
||||
listKey: "timer:job_listkey",
|
||||
zsetKey: "timer:once_zsetkey",
|
||||
listKey: "timer:once_listkey",
|
||||
redis: re,
|
||||
worker: w,
|
||||
}
|
||||
go wo.getTask()
|
||||
go wo.execTask()
|
||||
go wo.watch()
|
||||
})
|
||||
|
||||
return wo
|
||||
@@ -59,7 +71,7 @@ func InitWorker(ctx context.Context, re *redis.Client, w WorkerInterface) *worke
|
||||
|
||||
// 添加任务
|
||||
// 重复插入就代表覆盖
|
||||
func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration, data map[string]interface{}) error {
|
||||
func (w *worker) Add(uniqueKey string, jobType string, delayTime time.Duration, data interface{}) error {
|
||||
if delayTime.Abs() != delayTime {
|
||||
return fmt.Errorf("时间间隔不能为负数")
|
||||
}
|
||||
@@ -125,8 +137,8 @@ Loop:
|
||||
}
|
||||
}
|
||||
|
||||
// 执行任务
|
||||
func (w *worker) execTask() {
|
||||
// 监听任务
|
||||
func (w *worker) watch() {
|
||||
for {
|
||||
keys, err := w.redis.BLPop(w.ctx, time.Second*10, w.listKey).Result()
|
||||
if err != nil {
|
||||
@@ -134,26 +146,39 @@ func (w *worker) execTask() {
|
||||
continue
|
||||
}
|
||||
|
||||
go func() {
|
||||
s := strings.Split(keys[1], "[:]")
|
||||
go w.doTask(keys[1])
|
||||
|
||||
// 读取数据
|
||||
str, err := w.redis.Get(w.ctx, keys[1]).Result()
|
||||
if err != nil {
|
||||
fmt.Println("execJob err:", err)
|
||||
return
|
||||
}
|
||||
ed := extendData{}
|
||||
json.Unmarshal([]byte(str), &ed)
|
||||
|
||||
fmt.Println("开始时间：", time.Now().Format("2006-01-02 15:04:05"))
|
||||
code := w.worker.Worker(s[0], s[1], ed.Data)
|
||||
|
||||
if code == WorkerCodeAgain {
|
||||
// 重新放入队列
|
||||
fmt.Println("重入时间：", time.Now().Format("2006-01-02 15:04:05"))
|
||||
w.Add(s[0], s[1], ed.Delay, ed.Data)
|
||||
}
|
||||
}()
|
||||
}
|
||||
}
|
||||
|
||||
func (w *worker) doTask(key string) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
fmt.Println("timer:定时器出错", err)
|
||||
log.Println("errStack", string(debug.Stack()))
|
||||
}
|
||||
}()
|
||||
|
||||
s := strings.Split(key, "[:]")
|
||||
|
||||
// 读取数据
|
||||
str, err := w.redis.Get(w.ctx, key).Result()
|
||||
if err != nil {
|
||||
fmt.Println("execJob err:", err)
|
||||
return
|
||||
}
|
||||
ed := extendData{}
|
||||
json.Unmarshal([]byte(str), &ed)
|
||||
|
||||
fmt.Println("开始时间：", time.Now().Format("2006-01-02 15:04:05"))
|
||||
code, t := w.worker.Worker(s[0], s[1], ed.Data)
|
||||
|
||||
if code == WorkerCodeAgain {
|
||||
// 重新放入队列
|
||||
fmt.Println("重入时间：", time.Now().Format("2006-01-02 15:04:05"))
|
||||
if t != 0 && t == t.Abs() {
|
||||
ed.Delay = t
|
||||
}
|
||||
w.Add(s[0], s[1], ed.Delay, ed.Data)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,2 @@
|
||||
package timerx
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package timer
|
||||
package timerx
|
||||
|
||||
// 作者：黄新云
|
||||
|
||||
@@ -14,25 +14,25 @@ import (
|
||||
|
||||
// 定时器
|
||||
// 原理：每毫秒的时间触发
|
||||
// 单机版重复时间间隔定时器
|
||||
|
||||
// uuid -> timerStr
|
||||
var timerMap = make(map[string]*timerStr)
|
||||
var timerMapMux sync.Mutex
|
||||
|
||||
var timerCount int // 当前定时数目
|
||||
var onceLimit sync.Once // 实现单例
|
||||
var nextTime = time.Now() // 下一次执行的时间
|
||||
var timerCount int // 当前定时数目
|
||||
var onceLimit sync.Once // 实现单例
|
||||
|
||||
type ContextValueKey string // 定义context 传递的Key类型
|
||||
type Single struct{}
|
||||
|
||||
const (
|
||||
extendParamKey ContextValueKey = "extend_param"
|
||||
)
|
||||
var sin *Single = nil
|
||||
|
||||
// 定时器类
|
||||
func InitSingle(ctx context.Context) {
|
||||
func InitSingle(ctx context.Context) *Single {
|
||||
onceLimit.Do(func() {
|
||||
timer := time.NewTicker( time.Millisecond*200)
|
||||
sin = &Single{}
|
||||
|
||||
timer := time.NewTicker(time.Millisecond * 200)
|
||||
go func(ctx context.Context) {
|
||||
Loop:
|
||||
for {
|
||||
@@ -43,7 +43,7 @@ func InitSingle(ctx context.Context) {
|
||||
continue
|
||||
}
|
||||
// 迭代定时器
|
||||
iteratorTimer(ctx, t)
|
||||
sin.iterator(ctx, t)
|
||||
// fmt.Println("timer: 执行")
|
||||
case <-ctx.Done():
|
||||
// 跳出循环
|
||||
@@ -53,10 +53,17 @@ func InitSingle(ctx context.Context) {
|
||||
log.Println("timer: initend")
|
||||
}(ctx)
|
||||
})
|
||||
|
||||
return sin
|
||||
}
|
||||
|
||||
// 间隔定时器
|
||||
func AddTimer(space time.Duration, call callback, extend ExtendParams) (int, error) {
|
||||
// @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()
|
||||
|
||||
@@ -75,7 +82,7 @@ func AddTimer(space time.Duration, call callback, extend ExtendParams) (int, err
|
||||
SpaceTime: space,
|
||||
CanRunning: make(chan struct{}, 1),
|
||||
UniqueKey: "",
|
||||
Extend: extend,
|
||||
ExtendData: extend,
|
||||
}
|
||||
|
||||
timerMap[fmt.Sprintf("%d", timerCount)] = &t
|
||||
@@ -88,21 +95,15 @@ func AddTimer(space time.Duration, call callback, extend ExtendParams) (int, err
|
||||
return timerCount, nil
|
||||
}
|
||||
|
||||
// 添加需要定时的规则
|
||||
func AddToTimer(space time.Duration, call callback) int {
|
||||
extend := ExtendParams{}
|
||||
count, _ := AddTimer(space, call, extend)
|
||||
return count
|
||||
}
|
||||
|
||||
func DelToTimer(index string) {
|
||||
// 删除定时器
|
||||
func (s *Single) Del(index string) {
|
||||
timerMapMux.Lock()
|
||||
defer timerMapMux.Unlock()
|
||||
delete(timerMap, index)
|
||||
}
|
||||
|
||||
// 迭代定时器列表
|
||||
func iteratorTimer(ctx context.Context, nowTime time.Time) {
|
||||
func (s *Single) iterator(ctx context.Context, nowTime time.Time) {
|
||||
timerMapMux.Lock()
|
||||
defer timerMapMux.Unlock()
|
||||
|
||||
@@ -151,7 +152,7 @@ func iteratorTimer(ctx context.Context, nowTime time.Time) {
|
||||
}
|
||||
}()
|
||||
// fmt.Printf("timer: 准备执行 %v %v \n", k, v.Tag)
|
||||
timerAction(ctx, v.Callback, v.UniqueKey, v.Extend)
|
||||
s.doTask(ctx, v.Callback, v.UniqueKey, v.ExtendData)
|
||||
default:
|
||||
// fmt.Printf("timer: 已在执行 %v %v \n", k, v.Tag)
|
||||
return
|
||||
@@ -174,33 +175,14 @@ func iteratorTimer(ctx context.Context, nowTime time.Time) {
|
||||
// fmt.Println("timer: one finish")
|
||||
}
|
||||
|
||||
// 定义各个回调函数
|
||||
type callback func(ctx context.Context) bool
|
||||
|
||||
// 定时器操作类
|
||||
// 这里不应painc
|
||||
func timerAction(ctx context.Context, call callback, uniqueKey string, extend ExtendParams) bool {
|
||||
func (s *Single) doTask(ctx context.Context, call callback, uniqueKey string, extend interface{}) error {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
fmt.Println("timer:定时器出错", err)
|
||||
log.Println("errStack", string(debug.Stack()))
|
||||
}
|
||||
}()
|
||||
ctx, cancel := context.WithCancel(ctx)
|
||||
defer cancel()
|
||||
|
||||
// 附加数据
|
||||
ctx = context.WithValue(ctx, extendParamKey, extend)
|
||||
|
||||
return call(ctx)
|
||||
}
|
||||
|
||||
// 快捷方法
|
||||
func GetExtendParams(ctx context.Context) (*ExtendParams, error) {
|
||||
val := ctx.Value(extendParamKey)
|
||||
params, ok := val.(ExtendParams)
|
||||
if !ok {
|
||||
return nil, errors.New("没找到参数")
|
||||
}
|
||||
return ¶ms, nil
|
||||
return call(ctx, extend)
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
package timer_test
|
||||
package timerx_test
|
||||
|
||||
import "testing"
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
package timer
|
||||
package timerx
|
||||
|
||||
import "time"
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
type timerStr struct {
|
||||
Callback callback // 需要回调的方法
|
||||
@@ -9,10 +12,12 @@ type timerStr struct {
|
||||
NextTime time.Time // [删]下一次执行的时间
|
||||
SpaceTime time.Duration // 任务间隔时间
|
||||
UniqueKey string // 全局唯一键
|
||||
Extend ExtendParams // 附加参数
|
||||
ExtendData interface{} // 附加参数
|
||||
}
|
||||
|
||||
// 扩展参数
|
||||
type ExtendParams struct {
|
||||
Params map[string]interface{} // 带出去的参数
|
||||
}
|
||||
|
||||
var nextTime = time.Now() // 下一次执行的时间
|
||||
|
||||
|
||||
// 定义各个回调函数
|
||||
type callback func(ctx context.Context, extendData interface{}) error
|
||||
|
||||
Reference in New Issue
Block a user