怎么在golang中实现一个redis延时消息队列功能
怎么在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 }
看完上述内容是否对您有帮助呢?如果还想对相关知识有进一步的了解或阅读更多相关文章,请关注创新互联行业资讯频道,感谢您对创新互联的支持。
当前题目:怎么在golang中实现一个redis延时消息队列功能
当前路径:http://myzitong.com/article/piohdh.html