主頁 > 知識(shí)庫 > golang實(shí)現(xiàn)redis的延時(shí)消息隊(duì)列功能示例

golang實(shí)現(xiàn)redis的延時(shí)消息隊(duì)列功能示例

熱門標(biāo)簽:學(xué)海導(dǎo)航地圖標(biāo)注 高德地圖標(biāo)注口訣 地圖標(biāo)注的汽車標(biāo) 廣州呼叫中心外呼系統(tǒng) 西部云谷一期地圖標(biāo)注 南通如皋申請(qǐng)開通400電話 江西轉(zhuǎn)化率高的羿智云外呼系統(tǒng) 中國地圖標(biāo)注省會(huì)高清 浙江高速公路地圖標(biāo)注

前言

在學(xué)習(xí)過程中發(fā)現(xiàn)redis的zset還可以用來實(shí)現(xiàn)輕量級(jí)的延時(shí)消息隊(duì)列功能,雖然可靠性還有待提高,但是對(duì)于一些對(duì)數(shù)據(jù)可靠性要求不那么高的功能要求完全可以實(shí)現(xiàn)。本次主要采用了redis中zset中的zadd, zrangebyscore 和 zdel來實(shí)現(xiàn)一個(gè)小demo。

提前準(zhǔn)備 安裝redis, redis-go

因?yàn)橛玫氖莔acOS, 直接

$ brew install redis
$ go get github.com/garyburd/redigo/redis

又因?yàn)楸容^懶,生成任務(wù)的唯一id時(shí),直接采用了bson中的objectId,所以:

$ go get gopkg.in/mgo.v2/bson

唯一id不是必須有,但如果之后有實(shí)際應(yīng)用需要攜帶,便于查找相應(yīng)任務(wù)。

生產(chǎn)者

通過一個(gè)for循環(huán)生成10w個(gè)任務(wù), 每一個(gè)任務(wù)有不同的時(shí)間

func producer() {
 count := 0
 //生成100000個(gè)任務(wù)
 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函數(shù)在另一個(gè)包中, 將上一個(gè)函數(shù)中隨機(jī)生成的時(shí)間作為需要處理的時(shí)間戳.

// 添加任務(wù)
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))
}

消費(fèi)者

消費(fèi)者處理流程分為兩個(gè)步驟:

  • 獲取小于等于當(dāng)前時(shí)間戳的任務(wù)
  • 通過刪除當(dāng)前任務(wù)來判斷誰獲得了當(dāng)前任務(wù)

因?yàn)樵讷@取小于等于當(dāng)前時(shí)間戳的任務(wù)時(shí),可能有多個(gè)go routine同時(shí)讀到了當(dāng)前任務(wù),而只有一個(gè)任務(wù)可以來處理當(dāng)前任務(wù)。因此我們需要通過一個(gè)方案來判斷究竟由誰來處理這個(gè)任務(wù)(當(dāng)然如果只有一個(gè)消費(fèi)者可以讀到就直接處理):這個(gè)時(shí)候可以通過redis的刪除操作來獲取,因?yàn)閯h除指定value時(shí)只有成功的操作才會(huì)返回不為0,所以我們可以認(rèn)為刪除當(dāng)前隊(duì)列成功的那個(gè)go routine拿到了當(dāng)前的任務(wù)。

下面是代碼:

// 消費(fèi)者
func consumer() {
 // 啟動(dòng)10個(gè)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]
 // 如果當(dāng)前搶redis隊(duì)列成功,
 if redis.Client.DelJob(currentJob) > 0 {
  var jobMessage job.JobMessage
  util.JsonDecode(currentJob, jobMessage) //自定義的json解析函數(shù)
  handleMessage(jobMessage)
 }

 }

 }()
 count++
 }
}

// 處理任務(wù)用函數(shù)
func handleMessage(msg *job.JobMessage) {
 fmt.Printf("deal job: %s, require time: %d \n", msg.Id, msg.DealTime)
 go func() {
 countChan - true
 }()
}

redis部分的代碼,獲取任務(wù)和刪除任務(wù)

// 獲取任務(wù)
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
}

// 刪除當(dāng)前任務(wù), 用來判斷是否搶到了當(dāng)前任務(wù)
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
}

代碼大抵如此。最后跑起來之后,大概每3-4秒鐘能夠處理掉1w個(gè)任務(wù),速度上確實(shí)是...

以上就是本文的全部內(nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。

您可能感興趣的文章:
  • Java實(shí)現(xiàn)Redis延時(shí)消息隊(duì)列

標(biāo)簽:貴州 東營 常州 曲靖 吐魯番 許昌 德宏 保定

巨人網(wǎng)絡(luò)通訊聲明:本文標(biāo)題《golang實(shí)現(xiàn)redis的延時(shí)消息隊(duì)列功能示例》,本文關(guān)鍵詞  golang,實(shí)現(xiàn),redis,的,延時(shí),;如發(fā)現(xiàn)本文內(nèi)容存在版權(quán)問題,煩請(qǐng)?zhí)峁┫嚓P(guān)信息告之我們,我們將及時(shí)溝通與處理。本站內(nèi)容系統(tǒng)采集于網(wǎng)絡(luò),涉及言論、版權(quán)與本站無關(guān)。
  • 相關(guān)文章
  • 下面列出與本文章《golang實(shí)現(xiàn)redis的延時(shí)消息隊(duì)列功能示例》相關(guān)的同類信息!
  • 本頁收集關(guān)于golang實(shí)現(xiàn)redis的延時(shí)消息隊(duì)列功能示例的相關(guān)信息資訊供網(wǎng)民參考!
  • 推薦文章