怎么在golang中实现一个redis延时消息队列功能？很多新手对此不是很清楚，为了帮助大家解决这个难题，下面小编将为大家详细讲解，有这方面需求的人可以来学习下，希望你能有所收获。
提前准备 安装redis， redis-go
因为用的是macOS， 直接
计算
安全
数据库
网络和加速
企业服务
$ brew install redis
$ go get github.com/garyburd/redigo/redis又因为比较懒，生成任务的唯一id时，直接采用了bson中的objectId，所以：
$ go get gopkg.in/mgo.v2/bson唯一id不是必须有，但如果之后有实际应用需要携带，便于查找相应任务。
生产者
通过一个for循环生成10w个任务, 每一个任务有不同的时间
func producer() {
count := 0
//生成100000个任务
for count < 100000 {
count++
dealTime := int64(rand.Intn(5)) + time.Now().Unix()
uuid := bson.NewObjectId().Hex()
redis.Client.AddJob(&job.JobMessage{
Id: uuid,
DealTime: dealTime,
}, + int64(dealTime))
}
}其中AddJob函数在另一个包中, 将上一个函数中随机生成的时间作为需要处理的时间戳.
// 添加任务
func (client *RedisClient) AddJob(msg *job.JobMessage, dealTime int64) {
conn := client.Get()
defer conn.Close()
key := "JOB_MESSAGE_QUEUE"
conn.Do("zadd", key, dealTime, util.JsonEncode(msg))
}消费者
消费者处理流程分为两个步骤：
获取小于等于当前时间戳的任务
通过删除当前任务来判断谁获得了当前任务
因为在获取小于等于当前时间戳的任务时，可能有多个go routine同时读到了当前任务，而只有一个任务可以来处理当前任务。因此我们需要通过一个方案来判断究竟由谁来处理这个任务（当然如果只有一个消费者可以读到就直接处理）：这个时候可以通过redis的删除操作来获取，因为删除指定value时只有成功的操作才会返回不为0，所以我们可以认为删除当前队列成功的那个go routine拿到了当前的任务。
下面是代码：
// 消费者
func consumer() {
// 启动10个go routine一起去拿
count := 0
for count < 10 {
go func() {
for {
jobs := redis.Client.GetJob()
if len(jobs) <= 0 {
time.Sleep(time.Second * 1)
continue
}
currentJob := jobs[0]
// 如果当前抢redis队列成功，
if redis.Client.DelJob(currentJob) > 0 {
var jobMessage job.JobMessage
util.JsonDecode(currentJob, &jobMessage) //自定义的json解析函数
handleMessage(&jobMessage)
}
}
}()
count++
}
}
// 处理任务用函数
func handleMessage(msg *job.JobMessage) {
fmt.Printf("deal job: %s, require time: %d \n", msg.Id, msg.DealTime)
go func() {
countChan <- true
}()
}redis部分的代码，获取任务和删除任务
// 获取任务
func (client *RedisClient) GetJob() []string {
conn := client.Get()
defer conn.Close()
key := "JOB_MESSAGE_QUEUE"
timeNow := time.Now().Unix()
ret, err := redis.Strings(conn.Do("zrangebyscore", key, 0, timeNow, "limit", 0, 1))
if err != nil {
panic(err)
}
return ret
}
// 删除当前任务， 用来判断是否抢到了当前任务
func (client *RedisClient) DelJob(value string) int {
conn := client.Get()
defer conn.Close()
key := "JOB_MESSAGE_QUEUE"
ret, err := redis.Int(conn.Do("zrem", key, value))
if err != nil {
panic(err)
}
return ret
}看完上述内容是否对您有帮助呢？如果还想对相关知识有进一步的了解或阅读更多相关文章，请关注亿速云行业资讯频道，感谢您对亿速云的支持。
亿速云「云服务器」，即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘，价格低至29元/月。点击查看>>
免责声明：本站发布的内容（图片、视频和文字）以原创、转载和分享为主，文章观点不代表本网站立场，如果涉及侵权请联系站长邮箱：is@yisu.com进行举报，并提供相关证据，一经查实，将立刻删除涉嫌侵权内容。
