实际项目当中,经常会使用延迟队列,执行一些延迟任务,比如:
- 如果第一次通知失败,就延迟 X 分钟再通知一次,这在构建webhook的项目中很常见。
- 如果用户成功消费,X 天后没有评论,就默认给予好评。
这些任务具有的共同点可以抽象为:在某项任务执行完毕后,延迟一定时间执行另外一项任务。由于这些任务分布在随机的时间点上,构建定时JOB去跑这些任务必定是一件非常繁琐的工程。即使采用每次计算好执行时间,然后存储到缓存或者DB里面,仍然面临每次需要遍历庞大任务数据,并且执行时间判断造成的误差会导致某些任务也许无法执行的窘境。那么一个设计优良的延迟队列,为了解决上述问题,应该具备哪些特性呢?我觉得,它应该具备以下特性:
- 延迟时间要精准,保证在延迟到达时,执行预先设定的任务。
- 任务的执行时间,不应太过于影响时间计时器的运行。
- 延迟粒度可以自由配置,时间间隔可大可小。
要实现延迟队列,目前比较常用的方式,是使用时间轮算法,该算法的结构大体如下图所示:
![](/images/timewheel1.png)
它由三部分组成:
- 一个计时器,或者称为时间发生器。
- 一个循环队列,一般通过算法由数组来实现。
- 一个存储任务(task)的数据结构,可以是数组或者链表。
其中前两个决定了延迟的粒度,最后一个决定任务的存储方式;当任务执行完毕后,需要将任务从队列中删除,因此采用链表来存储任务,应该效率要比数组高很多(因为整个任务执行过程,不存在随机读取任务的需求)。
现在我们就用go语言来实现一个相对通用的延迟队列:
- 先来构建一个简单任务执行接口,具体的业务对象可以实现它来执行任务。
type Executor interface {
DoDelayTask(contents string) error
}
- 然后构建存储在时间轮中的任务,每个任务节点,我们采用链表实现。
//定义一个获取实现具体任务对象的工厂方法,该方法需要在业务项目中定义并实现
type BuildExecutor func(taskType string) Executor
//任务采用链表结构
type Task struct {
//任务在时间轮上的循环次数,等于0时,执行该任务
CycleCount int
//任务类型,用于工厂方法判断该使用哪个实现对象
TaskType string
//任务方法参数
TaskParams string
//任务工厂方法指针
TaskExecutor BuildExecutor
Next *Task
}
//通过工厂方法获取具体实现,然后调用方法,执行任务
func (t *Task) Execute() error {
if t.TaskExecutor != nil {
executor := t.TaskExecutor(t.TaskType)
if executor != nil {
return executor.DoDelayTask(t.TaskParams)
} else {
return errors.New("executor is nil")
}
} else {
return errors.New("task build executor is nil")
}
}
- 最后,用时间轮算法实现时间轮,并将任务通知与延迟队列结合。
const (
//轮的时间长度,目前设置为一个小时,也就是时间轮每循环一次需要1小时;默认时间轮上面的每走一步的最小粒度为1秒。
WHEEL_SIZE = 3600
)
//定义一个获取实现具体任务对象的工厂方法,该方法需要在业务项目中定义并实现
type BuildExecutor func(taskType string) Executor
var onceNew sync.Once
var onceStart sync.Once
var delayQueueInstance *delayQueue
type wheel struct {
//所有在时间轮上的任务均采用链表形式存储
//如果采用数组,对应已经执行过的任务,会造成不必要的空间浪费或者数组移动造成的时间复杂度
NotifyTasks *Task
}
type delayQueue struct {
//循环队列
TimeWheel [WHEEL_SIZE]wheel
CurrentIndex uint //时间轮当前指针
Persistence
//任务工厂方法指针, 需要在对象创建时初始化它
TaskExecutor BuildExecutor
}
//单列方法,使用默认的redis持久方案
func GetDelayQueue(serviceBuilder BuildExecutor) *delayQueue {
//保证只初始化一次
onceNew.Do(func() {
delayQueueInstance = &delayQueue{
Persistence: getRedisDb(),
TaskExecutor: serviceBuilder,
}
})
return delayQueueInstance
}
//单列方法,使用外部传入的持久方案
func GetDelayQueueWithPersis(serviceBuilder BuildExecutor, persistence Persistence) *delayQueue {
if persistence == nil {
log.Fatalf("persistance is null")
}
//保证只初始化一次
onceNew.Do(func() {
delayQueueInstance = &delayQueue{
Persistence: persistence,
TaskExecutor: serviceBuilder,
}
})
return delayQueueInstance
}
func (dq *delayQueue) Start() {
//保证只会有一个时间轮计时器
onceStart.Do(dq.init)
}
func (dq *delayQueue) init() {
go func() {
//从缓存中加载持久化的任务
dq.loadTasksFromDb()
for {
select {
//默认时间轮上的最小粒度为1秒
case <-time.After(time.Second * 1):
if dq.CurrentIndex >= WHEEL_SIZE {
dq.CurrentIndex = dq.CurrentIndex % WHEEL_SIZE
}
taskLinkHead := dq.TimeWheel[dq.CurrentIndex].NotifyTasks
//遍历链表
//当前节点前一指针
prev := taskLinkHead
//当前节点指针
p := taskLinkHead
for p != nil {
if p.CycleCount == 0 {
taskId := p.Id
//开启新的go routing 去做通知,加快每次遍历的速度,确保不会拖慢时间轮的运行
//如果任务有异常,尽量让具体的业务对象去处理,延迟队列不处理具体业务异常,
//这样可以保证延迟队列的业务单纯性,避免难以维护的问题。如果具体业务出现问题,需要重复通知,可以将任务重新加入队列即可。
go dq.ExecuteTask(p.TaskType, p.TaskParams)
//删除链表节点 task
//如果是第一个节点
if prev == p {
dq.TimeWheel[dq.CurrentIndex].NotifyTasks = p.Next
prev = p.Next
p = p.Next
} else {
//如果不是第一个节点
prev.Next = p.Next
p = p.Next
}
//从持久对象上删除该任务
dq.Persistence.Delete(taskId)
} else {
p.CycleCount--
prev = p
p = p.Next
}
}
dq.CurrentIndex++
}
}
}()
}
func (dq *delayQueue) loadTasksFromDb() {
tasks := dq.Persistence.GetList()
if tasks != nil && len(tasks) > 0 {
for _, task := range tasks {
// fmt.Printf("%v\n", task)
delaySeconds := ((task.CycleCount + 1) * WHEEL_SIZE) + task.WheelPosition
if delaySeconds > 0 {
dq.internalPush(time.Duration(delaySeconds)*time.Second, task.Id, task.TaskType, task.TaskParams, false)
}
}
}
}
//将任务加入延迟队列
func (dq *delayQueue) Push(delaySeconds time.Duration, taskType string, taskParams interface{}) error {
var pms string
result, ok := taskParams.(string)
if !ok {
tp, _ := json.Marshal(taskParams)
pms = string(tp)
} else {
pms = result
}
return dq.internalPush(delaySeconds, "", taskType, pms, true)
}
func (dq *delayQueue) internalPush(delaySeconds time.Duration, taskId string, taskType string, taskParams string, notNeedPresis bool) error {
if int(delaySeconds.Seconds()) == 0 {
errorMsg := fmt.Sprintf("the delay time cannot be less than 1 second, current is: %v", delaySeconds)
log.Println(errorMsg)
return errors.New(errorMsg)
}
//从当前时间指针处开始计时
calculateValue := int(dq.CurrentIndex) + int(delaySeconds.Seconds())
cycle := calculateValue / WHEEL_SIZE
if cycle > 0 {
cycle--
}
index := calculateValue % WHEEL_SIZE
if taskId == "" {
u := uuid.New()
taskId = u.String()
}
task := &Task{
Id: taskId,
CycleCount: cycle,
WheelPosition: index,
TaskType: taskType,
TaskParams: taskParams,
}
if dq.TimeWheel[index].NotifyTasks == nil {
dq.TimeWheel[index].NotifyTasks = task
// log.Println(dq.TimeWheel[index].NotifyTasks)
} else {
//将新任务插入链表头,由于任务之间没有顺序关系,这种实现最为简单
head := dq.TimeWheel[index].NotifyTasks
task.Next = head
dq.TimeWheel[index].NotifyTasks = task
// log.Println(dq.TimeWheel[index].NotifyTasks)
}
if notNeedPresis {
//持久化任务
dq.Persistence.Save(task)
}
return nil
}
//通过工厂方法获取具体实现,然后调用方法,执行任务
func (dq *delayQueue) ExecuteTask(taskType, taskParams string) error {
if dq.TaskExecutor != nil {
executor := dq.TaskExecutor(taskType)
if executor != nil {
log.Printf("Execute task: %s with params: %s\n", taskType, taskParams)
return executor.DoDelayTask(taskParams)
} else {
return errors.New("executor is nil")
}
} else {
return errors.New("task build executor is nil")
}
}
//延迟队列上,某个时间轮上的任务数量
func (dq *delayQueue) WheelTaskQuantity(index int) int {
tasks := dq.TimeWheel[index].NotifyTasks
if tasks == nil {
return 0
}
k := 0
for p := tasks; p != nil; p = p.Next {
k++
}
return k
}
上面我们用单列模式来构建延迟队列,因为很多时候,我们程序当中需要保证任何时刻只能有一个时间轮队列存在,不然会造成资源浪费。我们设置时间轮的大小和最小粒度分别为一个小时和一秒钟,这也是最为常用的一种方式。 用一个1秒的时间发生器配合select来驱动时间轮的运行,在每个轮节点上,遍历链表,查找需要执行的任务;任务执行时,无须处理异常,让具体业务对象去处理业务异常,保证时间轮的单纯性。
最后,我们写一个测试用例来测试时间轮的运行:
//创建一个测试业务对象,让它实现 Executor 接口
type businessNotify struct {
}
func (bn *businessNotify) DoDelayTask(contents string) error {
fmt.Println(fmt.Sprintf("Do task.....%s", contents))
return nil
}
//一个工厂方法,通过类型去构建任务执行对象
func commonFactory(taskType string) Executor {
//do filter business service by taskType ...
//...
return &businessNotify{}
}
//测试任务的执行
func TestTaskExecute(t *testing.T) {
task := &Task{
CycleCount: 1,
TaskType: "RetryNotify",
TaskParams: `{name: "raymond"}`,
TaskExecutor: commonFactory,
}
task.Execute()
}
//测试延迟队列的运行
func TestRunDelayQueue(t *testing.T) {
c := make(chan struct{})
queue := GetDelayQueue()
queue.Start()
go func() {
//模拟在其他地方将数据压入延迟队列
q := GetDelayQueue()
q.Push(5, "RetryNotify", `{name: "raymond"}`, commonFactory)
q.Push(15, "GoodComments", `{name: "raymond"}`, commonFactory)
q.Push(25, "TaskOne", `{name: "raymond"}`, commonFactory)
q.Push(35, "TaskThree", `{name: "raymond"}`, commonFactory)
q.Push(45, "TaskFour", `{name: "raymond"}`, commonFactory)
}()
c <- struct{}{}
}
现在各方面无误的话,可以看到时间轮正常运行,并在合适的时间执行对应的任务。
按理说一个简单的延迟队列就算实现完毕了,是不是我们就可以把它 run on production?且慢。。。一个真正可以在生产环境当中运行的延迟队列,还有很多事情需要考虑,比如:
- 任务是否需要持久化?如果不持久化,如何保证程序挂掉后,未执行的任务可以恢复执行?
- 采用何种持久化方案,更能适配当前的业务?
- 任务出错后,该采用什么补救方案(比如重试),才能保证任务执行的可靠性。
下次,我们将慢慢细化该延迟队列,将它打造成一个可以在生产环境稳定运行的程序 。。。