化妆网站模板下载免费,wordpress 外观自定义,购物网站开发背景需求,网站开发项目管理golang实现延迟队列
1 延迟队列#xff1a;邮件提醒、订单自动取消 延迟队列#xff1a;处理需要在未来某个特定时间执行的任务。这些任务被添加到队列中#xff0c;并且指定了一个执行时间#xff0c;只有达到指定的时间点时才能从队列中取出并执行。 应用场景#xff1…golang实现延迟队列
1 延迟队列邮件提醒、订单自动取消 延迟队列处理需要在未来某个特定时间执行的任务。这些任务被添加到队列中并且指定了一个执行时间只有达到指定的时间点时才能从队列中取出并执行。 应用场景 邮件提醒订单自动取消超过多少时间未支付就取消订单对超时任务的处理等 由于任务的执行是在未来的某个时间点因此这些任务不会立即执行而是存储在队列中直到它的预定执行时间才会被执行。
2 实现
2.1 simple简单版go自带的time包实现 思路 定义Task结构体包含 ExecuteTime time.TimeJob func() 定义DelayQueue TaskQueue []Taskfunc AddTaskfunc RemoveTaskExecuteTask 这种方案存在的问题 Go程序重启时存储在slice中的延迟处理任务将全部丢失 完整代码
package mainimport (fmttime
)/*
基于go实现延迟队列
*/
type Task struct {ExecuteTime time.TimeJob func()
}type DelayQueue struct {Tasks []*Task
}func (d *DelayQueue) AddTask(t *Task) {d.Tasks append(d.Tasks, t)
}func (d *DelayQueue) RemoveTask() {//FIFO: remove the first task to enqueued.Tasks d.Tasks[1:]
}func (d *DelayQueue) ExecuteTask() {for len(d.Tasks) 0 {//dequeue a taskcurrentTask : d.Tasks[0]if time.Now().Before(currentTask.ExecuteTime) {//if the task execution time is not up, waittime.Sleep(currentTask.ExecuteTime.Sub(time.Now()))}//execute the taskcurrentTask.Job()//remove task who has been executedd.RemoveTask()}}func main() {fmt.Println(start delayQueue)delayQueue : DelayQueue{}firstTask : Task{ExecuteTime: time.Now().Add(time.Second * 1),Job: func() {fmt.Println(executed task 1 after delay)},}delayQueue.AddTask(firstTask)secondTask : Task{ExecuteTime: time.Now().Add(time.Second * 7),Job: func() {fmt.Println(executed task 2 after delay)},}delayQueue.AddTask(secondTask)delayQueue.ExecuteTask()fmt.Println(all tasks have been done!!!)
}效果
2.2 complex持久版goredis 为了防止Go重启后存储到delayQueue的数据丢失我们可以将任务持久化到redis中。 思路 初始化redis连接延迟队列采用redis的zset有序集合实现 前置准备
# 安装docker
yum install -y yum-utils
yum-config-manager \--add-repo \https://download.docker.com/linux/centos/docker-ce.repo
yum install docker
systemctl start docker# docker搭建redis
mkdir -p /Users/ziyi2/docker-home/redis
docker run -d --name redis -v /Users/ziyi2/docker-home/redis:/data -p 6379:6379 redis完整代码
package mainimport (fmtgithub.com/go-redis/redislog github.com/ziyifast/logtime
)/*
基于redis zset实现延迟队列
*/
var redisdb *redis.Client
var DelayQueueKey delay-queuefunc initClient() (err error) {redisdb redis.NewClient(redis.Options{Addr: localhost:6379,Password: , // not set passwordDB: 0, //use default db})_, err redisdb.Ping().Result()if err ! nil {log.Errorf(%v, err)return err}return nil
}func main() {err : initClient()if err ! nil {log.Errorf(init redis client err: %v, err)return}addTaskToQueue(task1, time.Now().Add(time.Second*3).Unix())addTaskToQueue(task2, time.Now().Add(time.Second*8).Unix())//执行队列中的任务getAndExecuteTask()
}// executeTime为unix时间戳作为zset中的score。允许redis按照task应该执行时间来进行排序
func addTaskToQueue(task string, executeTime int64) {err : redisdb.ZAdd(DelayQueueKey, redis.Z{Score: float64(executeTime),Member: task,}).Err()if err ! nil {panic(err)}
}// 从redis中取一个task并执行
func getAndExecuteTask() {for {tasks, err : redisdb.ZRangeByScore(DelayQueueKey, redis.ZRangeBy{Min: -inf,Max: fmt.Sprintf(%d, time.Now().Unix()),Offset: 0,Count: 1,}).Result()if err ! nil {time.Sleep(time.Second * 1)continue}//处理任务for _, task : range tasks {fmt.Println(Execute task: , task)//执行完任务之后用 ZREM 移除该任务redisdb.ZRem(DelayQueueKey, task)}time.Sleep(time.Second * 1)}
}效果 redis一直从延迟队列中取数据如果处理完一批则睡眠1s 具体根据大家的业务调整此处主要介绍思路
