42 lines
901 B
Go
42 lines
901 B
Go
package task
|
|
|
|
import (
|
|
"ArmedPolice/config"
|
|
cache2 "ArmedPolice/serve/cache"
|
|
"ArmedPolice/serve/cache/logic"
|
|
"ArmedPolice/utils"
|
|
"fmt"
|
|
"time"
|
|
)
|
|
|
|
type TaskQueue struct{}
|
|
|
|
func (this *TaskQueue) Queue() {
|
|
go utils.TryCatch(func() {
|
|
for {
|
|
now := time.Now()
|
|
|
|
cache, _ := cache2.Cache.ZRangebyscore(config.RedisKeyForTaskQueue, &logic.ScoreRangeBy{Min: "0",
|
|
Max: fmt.Sprintf("%d", now.Unix())})
|
|
|
|
if len(cache) > 0 {
|
|
for _, v := range cache {
|
|
body, _ := cache2.Cache.HGet(config.RedisKeyForTaskQueueBody, v)
|
|
task := new(Task)
|
|
_ = task.UnmarshalBinary([]byte(body))
|
|
task.Consume()
|
|
}
|
|
// 销毁信息
|
|
_ = cache2.Cache.ZRem(config.RedisKeyForTaskQueue, cache)
|
|
_ = cache2.Cache.HDel(config.RedisKeyForTaskQueueBody, cache...)
|
|
}
|
|
// 每秒执行一次
|
|
time.Sleep(time.Second)
|
|
}
|
|
})
|
|
}
|
|
|
|
func NewTaskQueue() *TaskQueue {
|
|
return &TaskQueue{}
|
|
}
|