教程集 > Golang编程 > golang教程 > 正文 golang 实现延迟消息原理与方法

golang 实现延迟消息原理与方法

发布时间：2022-02-11 编辑：jiaochengji.com

教程集为您提供golang 实现延迟消息原理与方法等资源，欢迎您收藏本站，我们将为您提供最新的golang 实现延迟消息原理与方法资源

原文：https://www.cnblogs.com/jkko123/p/7239420.html

实现延迟消息最主要的两个结构：

环形队列：通过golang中的数组实现，分成3600个slot。

任务集合：通过map[key]*Task，每个slot一个map，map的值就是我们要执行的任务。

原理图如下：

<pre><code class="lang-go hljs">package main import ( "fmt" "errors" "time" ) type DelayMessage struct { curIndex int; slots [3600]map[string]*Task; closed chan bool; taskClose chan bool; timeClose chan bool; startTime time.Time; } type TaskFunc func(args ... interface{}) type Task struct { cycleNum int; exec TaskFunc; params []interface{}; } func NewDelayMessage() *DelayMessage{ dm := &DelayMessage{ curIndex: 0, closed:make(chan bool), taskClose: make(chan bool), timeClose: make(chan bool), startTime: time.Now(), } for i := 0; i < 3600; i { dm.slots[i] = make(map[string]*Task); } return dm; } func (dm *DelayMessage)Close(){ dm.closed <- true; } func (dm *DelayMessage)taskLoop(){ defer func() { fmt.Println("taskLoop exit") }(); for { select{ case <- dm.taskClose: { return } default: { tasks := dm.slots[dm.curIndex]; if len(tasks) > 0{ for k, v := range tasks{ if v.cycleNum == 0{ go v.exec(v.params...); delete(tasks, k); }else{ v.cycleNum--; } } } } } } } //启动延迟消息 func (dm *DelayMessage) Start() { go dm.taskLoop(); go dm.timeLoop(); select { case <-dm.closed: { dm.taskClose <- true; dm.timeClose <- true; break; } }; } //处理每1秒移动下标 func (dm *DelayMessage) timeLoop() { defer func() { fmt.Println("timeLoop exit"); }(); tick := time.NewTicker(time.Second); for { select { case <-dm.timeClose: { return; } case <-tick.C: { fmt.Println(time.Now().Format("2006-01-02 15:04:05")); //判断当前下标，如果等于3599则重置为0，否则加1 if dm.curIndex == 3599 { dm.curIndex = 0; } else { dm.curIndex ; } } } } } //添加任务 func (dm *DelayMessage) AddTask(t time.Time, key string, exec TaskFunc, params []interface{}) error { if dm.startTime.After(t) { return errors.New("时间错误"); } //当前时间与指定时间相差秒数 subSecond := t.Unix() - dm.startTime.Unix(); //计算循环次数 cycleNum := int(subSecond / 3600); //计算任务所在的slots的下标 ix := subSecond % 3600; //把任务加入tasks中 tasks := dm.slots[ix]; if _, ok := tasks[key]; ok { return errors.New("该slots中已存在key为" key "的任务"); } tasks[key] = &Task{ cycleNum: cycleNum, exec: exec, params: params, }; return nil; } func main() { //创建延迟消息 dm := NewDelayMessage(); //添加任务 dm.AddTask(time.Now().Add(time.Second*10), "test1", func(args ...interface{}) { fmt.Println(args...); }, []interface{}{1, 2, 3}); dm.AddTask(time.Now().Add(time.Second*10), "test2", func(args ...interface{}) { fmt.Println(args...); }, []interface{}{4, 5, 6}); dm.AddTask(time.Now().Add(time.Second*20), "test3", func(args ...interface{}) { fmt.Println(args...); }, []interface{}{"hello", "world", "test"}); dm.AddTask(time.Now().Add(time.Second*30), "test4", func(args ...interface{}) { sum := 0; for arg := range args { sum = arg; } fmt.Println("sum : ", sum); }, []interface{}{1, 2, 3}); //40秒后关闭 time.AfterFunc(time.Second*40, func() { dm.Close(); }); dm.Start(); } </code></pre>

<blockquote class="layui-elem-quote" style="width: 100%;overflow:hidden"> 作者： zhizhengguan
链接： https://blog.csdn.net/zhizhengguan/article/details/90576511?ops_request_misc=%7B%22request%5Fid%22%3A%22161130781216780262544451%22%2C%22scm%22%3A%2220140713.130102334.pc%5Fblog.%22%7D&request_id=161130781216780262544451&biz_
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。
</blockquote>

到此这篇关于“golang 实现延迟消息原理与方法”的文章就介绍到这了,更多文章或继续浏览下面的相关文章，希望大家以后多多支持JQ教程网！

您可能感兴趣的文章：
订阅redis接收的json字符串斜杠_用 Golang 实现基于 Redis 的安全高效 RPC 通信
 go 关键字之 defer
go 学习笔记之咬文嚼字带你弄清楚 defer 延迟函数
 图解 Go 中的延迟调用 defer
如何在生产环境mysql删除亿万级数据解并且不影响数据库主从延迟的解决方案
 Go语言的实时GC——理论与实践
 订单系统设计 —— 数据同步与监控
 Golang defer 你不知道的事
 Go：defer 语句如何工作
 使用 Go 重构 - Goroutine 并发

上一篇：golang读取接口数据interface{} 下一篇：golang函数后的 {

[关闭]

golang 实现延迟消息原理与方法

最近更新

浏览排行