免费资源部落

这篇文章主要讲解了“golang如何实现延时任务”，文中的讲解内容简单清晰，易于学习与理解，下面请大家跟着小编的思路慢慢深入，一起来研究和学习“golang如何实现延时任务”吧！

实现思路

我们都知道，任何一种队列，实际上都是存在生产者和消费者两部分的。只不过，延时任务相对于普通队列，多了一个延时的特性罢了。

1、生产者

从生产者的角度上讲，当用户推送一个任务过来的时候，会携带着延迟执行的时间数值。为了让这个任务到预定时刻能执行，我们需要将这个任务放在内存里储存一段时间，并且时间是一维的，在不断增长。那么，我们用什么数据结构存储呢？

（1）选择一：map。由于map具有无序性，无法按照执行时间排序，我们无法保证取出的任务是否是当前时间点需要执行的，所以排除这个选项。

（2）选择二：channel。的确，channel有时候可以看作队列，然而，它的输出和输入严格遵循着“先进先出”的原则，遗憾的是，先进的任务未必就是先执行的，因此，channel也并不合适。

（3）选择三：slice。切片貌似可行，因为切片元素是具有有序性的，所以，如果我们能够按照执行时间的顺序排列好所有的切片元素，那么，每次只要读取切片的头元素（也可能是尾元素），就可以得到我们要的任务。

2、消费者

从消费者的角度来说，它最大的难点在于，如何让每个任务，在特定的时间点被消费。那么，针对每一个任务，我们如何实现，让它等待一段时间后再执行呢？

没错，就是timer。

总结下来，“切片+timer”的组合，应该是可以达到目的的。

步步为营

1、数据流

（1）用户调用InitDelayQueue() ，初始化延时任务对象。

（2）开启协程，监听任务操作管道（add/delete信号），以及执行时间管道（timer.C信号）。

（3）用户发出add/delete信号。

（4）（2）中的协程捕捉到（3）中的信号，对任务列表进行变更。

（5）当任务执行的时间点到达的时候（timer.C管道有元素输出的时候），执行任务。

2、数据结构

（1）延时任务对象

//延时任务对象typeDelayQueuestruct{tasks[]*task//存储任务列表的切片addchan*task//用户添加任务的管道信号removechanstring//用户删除任务的管道信号waitRemoveTaskMappingmap[string]struct{}//等待删除的任务id列表}

这里需要注意，有一个waitRemoveTaskMapping字段。由于要删除的任务，可能还在add管道中，没有及时更新到tasks字段中，所以，需要临时记录下客户要删除的任务id。

（2）任务对象

//任务对象typetaskstruct{idstring//任务idexecTimetime.Time//执行时间ffunc()//执行函数}

3、初始化延时任务对象

//初始化延时任务对象funcInitDelayQueue()*DelayQueue{q:=&DelayQueue{add:make(chan*task,10000),remove:make(chanstring,100),waitRemoveTaskMapping:make(map[string]struct{}),}returnq}

在这个过程中，我们需要对用户对任务的操作信号，以及任务的执行时间信号进行监听。

func(q*DelayQueue)start(){for{//todosomething...select{casenow:=<-timer.C://任务执行时间信号//todosomething...caset:=<-q.add://任务推送信号//todosomething...caseid:=<-q.remove://任务删除信号//todosomething...}}}

完善我们的初始化方法：

//初始化延时任务对象funcInitDelayQueue()*DelayQueue{q:=&DelayQueue{add:make(chan*task,10000),remove:make(chanstring,100),waitRemoveTaskMapping:make(map[string]struct{}),}//开启协程，监听任务相关信号goq.start()returnq}

4、生产者推送任务

生产者推送任务的时候，只需要将任务加到add管道中即可，在这里，我们生成一个任务id，并返回给用户。

//用户推送任务func(q*DelayQueue)Push(timeIntervaltime.Duration,ffunc())string{//生成一个任务id，方便删除使用id:=genTaskId()t:=&task{id:id,execTime:time.Now().Add(timeInterval),f:f,}//将任务推到add管道中q.add<-treturnid}

5、任务推送信号的处理

在这里，我们要将用户推送的任务放到延时任务的tasks字段中。由于，我们需要将任务按照执行时间顺序排序，所以，我们需要找到新增任务在切片中的插入位置。又因为，插入之前的任务列表已经是有序的，所以，我们可以采用二分法处理。

//使用二分法判断新增任务的插入位置func(q*DelayQueue)getTaskInsertIndex(t*task,leftIndex,rightIndexint)(indexint){iflen(q.tasks)==0{return}length:=rightIndex-leftIndexifq.tasks[leftIndex].execTime.Sub(t.execTime)>=0{//如果当前切片中最小的元素都超过了插入的优先级，则插入位置应该是最左边returnleftIndex}ifq.tasks[rightIndex].execTime.Sub(t.execTime)<=0{//如果当前切片中最大的元素都没超过插入的优先级，则插入位置应该是最右边returnrightIndex+1}iflength==1&&q.tasks[leftIndex].execTime.Before(t.execTime)&&q.tasks[rightIndex].execTime.Sub(t.execTime)>=0{//如果插入的优先级刚好在仅有的两个优先级之间，则中间的位置就是插入位置returnleftIndex+1}middleVal:=q.tasks[leftIndex+length/2].execTime//这里用二分法递归的方式，一直寻找正确的插入位置ift.execTime.Sub(middleVal)<=0{returnq.getTaskInsertIndex(t,leftIndex,leftIndex+length/2)}else{returnq.getTaskInsertIndex(t,leftIndex+length/2,rightIndex)}}

找到正确的插入位置后，我们才能将任务准确插入：

//将任务添加到任务切片列表中func(q*DelayQueue)addTask(t*task){//寻找新增任务的插入位置insertIndex:=q.getTaskInsertIndex(t,0,len(q.tasks)-1)//找到了插入位置，更新任务列表q.tasks=append(q.tasks,&task{})copy(q.tasks[insertIndex+1:],q.tasks[insertIndex:])q.tasks[insertIndex]=t}

那么，在监听add管道的时候，我们直接调用上述addTask() 即可。

func(q*DelayQueue)start(){for{//todosomething...select{casenow:=<-timer.C://任务执行时间信号//todosomething...caset:=<-q.add://任务推送信号q.addTask(t)caseid:=<-q.remove://任务删除信号//todosomething...}}}

6、生产者删除任务

//用户删除任务func(q*DelayQueue)Delete(idstring){q.remove<-id}

7、任务删除信号的处理

在这里，我们可以遍历任务列表，根据删除任务的id找到其在切片中的对应index。

//删除指定任务func(q*DelayQueue)deleteTask(idstring){deleteIndex:=-1forindex,t:=rangeq.tasks{ift.id==id{//找到了在切片中需要删除的所以呢deleteIndex=indexbreak}}ifdeleteIndex==-1{//如果没有找到删除的任务，说明任务还在add管道中，来不及更新到tasks中，这里我们就将这个删除id临时记录下来//注意，这里暂时不考虑，任务id非法的特殊情况q.waitRemoveTaskMapping[id]=struct{}{}return}iflen(q.tasks)==1{//删除后，任务列表就没有任务了q.tasks=[]*task{}return}ifdeleteIndex==len(q.tasks)-1{//如果删除的是，任务列表的最后一个元素，则执行下列代码q.tasks=q.tasks[:len(q.tasks)-1]return}//如果删除的是，任务列表的其他元素，则需要将deleteIndex之后的元素，全部向前挪动一位copy(q.tasks[deleteIndex:len(q.tasks)-1],q.tasks[deleteIndex+1:len(q.tasks)-1])q.tasks=q.tasks[:len(q.tasks)-1]return}

然后，我们可以完善start()方法了。

func(q*DelayQueue)start(){for{//todosomething...select{casenow:=<-timer.C://任务执行时间信号//todosomething...caset:=<-q.add://任务推送信号q.addTask(t)caseid:=<-q.remove://任务删除信号q.deleteTask(id)}}}

8、任务执行信号的处理

start()执行的时候，分成两种情况：任务列表为空，只需要监听add管道即可；任务列表不为空的时候，需要监听所有管道。任务执行信号，主要是依靠timer来实现，属于第二种情况。

func(q*DelayQueue)start(){for{iflen(q.tasks)==0{//任务列表为空的时候，只需要监听add管道select{caset:=<-q.add://添加任务q.addTask(t)}continue}//任务列表不为空的时候，需要监听所有管道//任务的等待时间=任务的执行时间-当前的时间currentTask:=q.tasks[0]timer:=time.NewTimer(currentTask.execTime.Sub(time.Now()))select{casenow:=<-timer.C://任务执行信号timer.Stop()if_,isRemove:=q.waitRemoveTaskMapping[currentTask.id];isRemove{//之前客户已经发出过该任务的删除信号，因此需要结束任务，刷新任务列表q.endTask()delete(q.waitRemoveTaskMapping,currentTask.id)continue}//开启协程，异步执行任务goq.execTask(currentTask,now)//任务结束，刷新任务列表q.endTask()caset:=<-q.add://任务推送信号timer.Stop()q.addTask(t)caseid:=<-q.remove://任务删除信号timer.Stop()q.deleteTask(id)}}}

执行任务：

//执行任务func(q*DelayQueue)execTask(task*task,currentTimetime.Time){iftask.execTime.After(currentTime){//如果当前任务的执行时间落后于当前时间，则不执行return}//执行任务task.f()return}

结束任务，刷新任务列表：

//一个任务去执行了，刷新任务列表func(q*DelayQueue)endTask(){iflen(q.tasks)==1{q.tasks=[]*task{}return}q.tasks=q.tasks[1:]}

9、完整代码

delay_queue.go

packagedelay_queueimport("go.mongodb.org/mongo-driver/bson/primitive""time")//延时任务对象typeDelayQueuestruct{tasks[]*task//存储任务列表的切片addchan*task//用户添加任务的管道信号removechanstring//用户删除任务的管道信号waitRemoveTaskMappingmap[string]struct{}//等待删除的任务id列表}//任务对象typetaskstruct{idstring//任务idexecTimetime.Time//执行时间ffunc()//执行函数}//初始化延时任务对象funcInitDelayQueue()*DelayQueue{q:=&DelayQueue{add:make(chan*task,10000),remove:make(chanstring,100),waitRemoveTaskMapping:make(map[string]struct{}),}//开启协程，监听任务相关信号goq.start()returnq}//用户删除任务func(q*DelayQueue)Delete(idstring){q.remove<-id}//用户推送任务func(q*DelayQueue)Push(timeIntervaltime.Duration,ffunc())string{//生成一个任务id，方便删除使用id:=genTaskId()t:=&task{id:id,execTime:time.Now().Add(timeInterval),f:f,}//将任务推到add管道中q.add<-treturnid}//监听各种任务相关信号func(q*DelayQueue)start(){for{iflen(q.tasks)==0{//任务列表为空的时候，只需要监听add管道select{caset:=<-q.add://添加任务q.addTask(t)}continue}//任务列表不为空的时候，需要监听所有管道//任务的等待时间=任务的执行时间-当前的时间currentTask:=q.tasks[0]timer:=time.NewTimer(currentTask.execTime.Sub(time.Now()))select{casenow:=<-timer.C:timer.Stop()if_,isRemove:=q.waitRemoveTaskMapping[currentTask.id];isRemove{//之前客户已经发出过该任务的删除信号，因此需要结束任务，刷新任务列表q.endTask()delete(q.waitRemoveTaskMapping,currentTask.id)continue}//开启协程，异步执行任务goq.execTask(currentTask,now)//任务结束，刷新任务列表q.endTask()caset:=<-q.add://添加任务timer.Stop()q.addTask(t)caseid:=<-q.remove://删除任务timer.Stop()q.deleteTask(id)}}}//执行任务func(q*DelayQueue)execTask(task*task,currentTimetime.Time){iftask.execTime.After(currentTime){//如果当前任务的执行时间落后于当前时间，则不执行return}//执行任务task.f()return}//一个任务去执行了，刷新任务列表func(q*DelayQueue)endTask(){iflen(q.tasks)==1{q.tasks=[]*task{}return}q.tasks=q.tasks[1:]}//将任务添加到任务切片列表中func(q*DelayQueue)addTask(t*task){//寻找新增任务的插入位置insertIndex:=q.getTaskInsertIndex(t,0,len(q.tasks)-1)//找到了插入位置，更新任务列表q.tasks=append(q.tasks,&task{})copy(q.tasks[insertIndex+1:],q.tasks[insertIndex:])q.tasks[insertIndex]=t}//删除指定任务func(q*DelayQueue)deleteTask(idstring){deleteIndex:=-1forindex,t:=rangeq.tasks{ift.id==id{//找到了在切片中需要删除的所以呢deleteIndex=indexbreak}}ifdeleteIndex==-1{//如果没有找到删除的任务，说明任务还在add管道中，来不及更新到tasks中，这里我们就将这个删除id临时记录下来//注意，这里暂时不考虑，任务id非法的特殊情况q.waitRemoveTaskMapping[id]=struct{}{}return}iflen(q.tasks)==1{//删除后，任务列表就没有任务了q.tasks=[]*task{}return}ifdeleteIndex==len(q.tasks)-1{//如果删除的是，任务列表的最后一个元素，则执行下列代码q.tasks=q.tasks[:len(q.tasks)-1]return}//如果删除的是，任务列表的其他元素，则需要将deleteIndex之后的元素，全部向前挪动一位copy(q.tasks[deleteIndex:len(q.tasks)-1],q.tasks[deleteIndex+1:len(q.tasks)-1])q.tasks=q.tasks[:len(q.tasks)-1]return}//寻找任务的插入位置func(q*DelayQueue)getTaskInsertIndex(t*task,leftIndex,rightIndexint)(indexint){//使用二分法判断新增任务的插入位置iflen(q.tasks)==0{return}length:=rightIndex-leftIndexifq.tasks[leftIndex].execTime.Sub(t.execTime)>=0{//如果当前切片中最小的元素都超过了插入的优先级，则插入位置应该是最左边returnleftIndex}ifq.tasks[rightIndex].execTime.Sub(t.execTime)<=0{//如果当前切片中最大的元素都没超过插入的优先级，则插入位置应该是最右边returnrightIndex+1}iflength==1&&q.tasks[leftIndex].execTime.Before(t.execTime)&&q.tasks[rightIndex].execTime.Sub(t.execTime)>=0{//如果插入的优先级刚好在仅有的两个优先级之间，则中间的位置就是插入位置returnleftIndex+1}middleVal:=q.tasks[leftIndex+length/2].execTime//这里用二分法递归的方式，一直寻找正确的插入位置ift.execTime.Sub(middleVal)<=0{returnq.getTaskInsertIndex(t,leftIndex,leftIndex+length/2)}else{returnq.getTaskInsertIndex(t,leftIndex+length/2,rightIndex)}}funcgenTaskId()string{returnprimitive.NewObjectID().Hex()}

测试代码：delay_queue_test.go

packagedelay_queueimport("fmt""testing""time")funcTestDelayQueue(t*testing.T){q:=InitDelayQueue()fori:=0;i<100;i++{gofunc(iint){id:=q.Push(time.Duration(i)*time.Second,func(){fmt.Printf("%d秒后执行...\n",i)return})ifi%7==0{q.Delete(id)}}(i)}time.Sleep(time.Hour)}

头脑风暴

上面的方案，的确实现了延时任务的效果，但是其中仍然有一些问题，仍然值得我们思考和优化。

1、按照上面的方案，如果大量延时任务的执行时间，集中在同一个时间点，会造成短时间内timer频繁地创建和销毁。

2、上述方案相比于time.AfterFunc()方法，我们需要在哪些场景下作出取舍。

3、如果服务崩溃或重启，如何去持久化队列中的任务。

感谢各位的阅读，以上就是“golang如何实现延时任务”的内容了，经过本文的学习后，相信大家对golang如何实现延时任务这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是，小编将为大家推送更多相关知识点的文章，欢迎关注！