实现延迟消息具体思路我是看的下面这篇文章
https://mp.weixin.qq.com/s/eDMV25YqCPYjxQG-dvqSqQ
实现延迟消息最主要的两个结构：
环形队列：通过golang中的数组实现，分成3600个slot。
任务集合：通过map[key]*Task，每个slot一个map，map的值就是我们要执行的任务。
原理图如下：
实现代码如下：
package main;
import (
"time"
"errors"
"fmt"
)
//延迟消息
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) Start() {
go dm.taskLoop();
go dm.timeLoop();
select {
case
{
dm.taskClose
dm.timeClose
break;
}
};
}
//关闭延迟消息
func (dm *DelayMessage) Close() {
dm.closed
}
//处理每1秒的任务
func (dm *DelayMessage) taskLoop() {
defer func() {
fmt.Println("taskLoop exit");
}();
for {
select {
case
{
return;
}
default:
{
//取出当前的槽的任务
tasks := dm.slots[dm.curIndex];
if len(tasks) > 0 {
//遍历任务，判断任务循环次数等于0，则运行任务
//否则任务循环次数减1
for k, v := range tasks {
if v.cycleNum == 0 {
go v.exec(v.params...);
//删除运行过的任务
delete(tasks, k);
} else {
v.cycleNum--;
}
}
}
}
}
}
}
//处理每1秒移动下标
func (dm *DelayMessage) timeLoop() {
defer func() {
fmt.Println("timeLoop exit");
}();
tick := time.NewTicker(time.Second);
for {
select {
case
{
return;
}
case
{
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();
}
测试结果如下：
