Go+Redis實(shí)現(xiàn)延遲隊(duì)列實(shí)操
前言
延遲隊(duì)列是一種非常使用的數(shù)據(jù)結(jié)構(gòu),我們經(jīng)常有需要延遲推送處理消息的場景,比如延遲60秒發(fā)送短信,延遲30分鐘關(guān)閉訂單,消息消費(fèi)失敗延遲重試等等。
一般我們實(shí)現(xiàn)延遲消息都需要依賴底層的有序結(jié)構(gòu),比如堆,而Redis剛好提供了zset
這種數(shù)據(jù)類型,它的底層實(shí)現(xiàn)是哈希表+跳表
,也是一種有序的結(jié)構(gòu),所以這篇文章主要是使用Go+Redis來實(shí)現(xiàn)延遲隊(duì)列。
當(dāng)然Redis本身并不支持延遲隊(duì)列,所以我們只是實(shí)現(xiàn)一個(gè)比較簡單的延遲隊(duì)列,而且Redis不太適合大量消息堆積,所以只適合比較簡單的場景,如果需要更加強(qiáng)大穩(wěn)定的消息隊(duì)列,可以使用RocketMQ
等自帶延遲消息的消息隊(duì)列。
我們這里先定一下我們要實(shí)現(xiàn)的幾個(gè)目標(biāo):
- 消息必須至少被消費(fèi)一次
- 多個(gè)生產(chǎn)者
- 多個(gè)消費(fèi)者
然后我們定義一個(gè)簡單的接口:
- Push(msg) error:添加消息到隊(duì)列
- Consume(topic, batchSize, func(msg) error):消費(fèi)消息
簡單的實(shí)現(xiàn)
- 每個(gè)主題最多可以被一個(gè)消費(fèi)者消費(fèi),因?yàn)椴粫?huì)對主題進(jìn)行分區(qū)
- 但是可以多個(gè)生產(chǎn)者同時(shí)進(jìn)行生產(chǎn),因?yàn)镻ush操作是原子的
- 同時(shí)需要消費(fèi)操作返回值error為nil才刪除消息,保證消息至少被消費(fèi)一次
定義消息
這個(gè)消息參考了Kafka的消息結(jié)構(gòu):
- Topic可以是某個(gè)隊(duì)列的名字
- Key是消息的唯一標(biāo)識(shí),在一個(gè)隊(duì)列里面不可以重復(fù)
- Body是消息的內(nèi)容
- Delay是消息的延遲時(shí)間
- ReadyTime是消息準(zhǔn)備好執(zhí)行的時(shí)間
// Msg 消息 type Msg struct { Topic string // 消息的主題 Key string // 消息的Key Body []byte // 消息的Body Delay time.Duration // 延遲時(shí)間(秒) ReadyTime time.Time // 消息準(zhǔn)備好執(zhí)行的時(shí)間(now + delay) }
Push
由于我們需要把消息的Body存儲(chǔ)到Hash
,把消息的ReadyTime存儲(chǔ)到ZSet
,所以我們需要一個(gè)簡單的Lua腳本來保證這兩個(gè)操作是原子的
。
同時(shí)我們不會(huì)覆蓋已經(jīng)存在的相同Key的消息。
const delayQueuePushRedisScript = ` -- KEYS[1]: topicZSet -- KEYS[2]: topicHash -- ARGV[1]: 消息的Key -- ARGV[2]: 消息的Body -- ARGV[3]: 消息準(zhǔn)備好執(zhí)行的時(shí)間 local topicZSet = KEYS[1] local topicHash = KEYS[2] local key = ARGV[1] local body = ARGV[2] local readyTime = tonumber(ARGV[3]) -- 添加readyTime到zset local count = redis.call("zadd", topicZSet, readyTime, key) -- 消息已經(jīng)存在 if count == 0 then return 0 end -- 添加body到hash redis.call("hsetnx", topicHash, key, body) return 1 `
func (q *SimpleRedisDelayQueue) Push(ctx context.Context, msg *Msg) error { // 如果設(shè)置了ReadyTime,就使用RedisTime var readyTime int64 if !msg.ReadyTime.IsZero() { readyTime = msg.ReadyTime.Unix() } else { // 否則使用Delay readyTime = time.Now().Add(msg.Delay).Unix() } success, err := q.pushScript.Run(ctx, q.client, []string{q.topicZSet(msg.Topic), q.topicHash(msg.Topic)}, msg.Key, msg.Body, readyTime).Bool() if err != nil { return err } if !success { return ErrDuplicateMessage } return nil }
Consume
其中第二個(gè)參數(shù)batchSize
表示用于批量獲取已經(jīng)準(zhǔn)備好執(zhí)行的消息,減少網(wǎng)絡(luò)請求。
fn
是對消息進(jìn)行處理的函數(shù),它有一個(gè)返回值error
,如果是nil
才表示消息消費(fèi)成功,然后調(diào)用刪除腳本把成功消費(fèi)的消息給刪除(需要原子的刪除ZSet和Hash里面的內(nèi)容)。
const delayQueueDelRedisScript = ` -- KEYS[1]: topicZSet -- KEYS[2]: topicHash -- ARGV[1]: 消息的Key local topicZSet = KEYS[1] local topicHash = KEYS[2] local key = ARGV[1] -- 刪除zset和hash關(guān)于這條消息的內(nèi)容 redis.call("zrem", topicZSet, key) redis.call("hdel", topicHash, key) return 1 `
func (q *SimpleRedisDelayQueue) Consume(topic string, batchSize int, fn func(msg *Msg) error) { for { // 批量獲取已經(jīng)準(zhǔn)備好執(zhí)行的消息 now := time.Now().Unix() zs, err := q.client.ZRangeByScoreWithScores(context.Background(), q.topicZSet(topic), &redis.ZRangeBy{ Min: "-inf", Max: strconv.Itoa(int(now)), Count: int64(batchSize), }).Result() // 如果獲取出錯(cuò)或者獲取不到消息,則休眠一秒 if err != nil || len(zs) == 0 { time.Sleep(time.Second) continue } // 遍歷每個(gè)消息 for _, z := range zs { key := z.Member.(string) // 獲取消息的body body, err := q.client.HGet(context.Background(), q.topicHash(topic), key).Bytes() if err != nil { continue } // 處理消息 err = fn(&Msg{ Topic: topic, Key: key, Body: body, ReadyTime: time.Unix(int64(z.Score), 0), }) if err != nil { continue } // 如果消息處理成功,刪除消息 q.delScript.Run(context.Background(), q.client, []string{q.topicZSet(topic), q.topicHash(topic)}, key) } } }
存在的問題
如果多個(gè)線程同時(shí)調(diào)用Consume
函數(shù),那么多個(gè)線程會(huì)拉取相同的可執(zhí)行的消息,造成消息重復(fù)的被消費(fèi)。
多消費(fèi)者實(shí)現(xiàn)
- 每個(gè)主題最多可以被分區(qū)個(gè)數(shù)個(gè)消費(fèi)者消費(fèi),會(huì)對主題進(jìn)行分區(qū)
定義消息
- 我們添加了一個(gè)Partition字段表示消息的分區(qū)號(hào)
// Msg 消息 type Msg struct { Topic string // 消息的主題 Key string // 消息的Key Body []byte // 消息的Body Partition int // 分區(qū)號(hào) Delay time.Duration // 延遲時(shí)間(秒) ReadyTime time.Time // 消息準(zhǔn)備好執(zhí)行的時(shí)間 }
Push
代碼與SimpleRedisDelayQueue的Push相同,只是我們會(huì)使用Msg里面的Partition字段對主題進(jìn)行分區(qū)。
func (q *PartitionRedisDelayQueue) topicZSet(topic string, partition int) string { return fmt.Sprintf("%s:%d:z", topic, partition) } func (q *PartitionRedisDelayQueue) topicHash(topic string, partition int) string { return fmt.Sprintf("%s:%d:h", topic, partition) }
Consume
代碼與SimpleRedisDelayQueue的Consume相同,我們也只是對Consume多加了一個(gè)partition
參數(shù)用于指定消費(fèi)的分區(qū)。
func (q *PartitionRedisDelayQueue) Consume(topic string, batchSize, partition int, fn func(msg *Msg) error) { // ... }
存在的問題
一個(gè)比較大的問題就是我們需要手動(dòng)指定分區(qū)而不是自動(dòng)分配分區(qū),這個(gè)問題對于Push操作解決起來比較容易,可以通過哈希算法對Key進(jìn)行哈希取模進(jìn)行分區(qū),比如murmur3。但是對于Consume就比較復(fù)雜,因?yàn)槲覀儽仨氂涗浤膫€(gè)分區(qū)已經(jīng)被消費(fèi)者消費(fèi)了。如果真的需要更加復(fù)雜的場景還是建議使用RocketMQ
、Kafka
等消息隊(duì)列進(jìn)行實(shí)現(xiàn)。
總結(jié)
- 使用Redis的ZSet可以很容易的實(shí)現(xiàn)一個(gè)高性能消息隊(duì)列
- 但是Redis的ZSet實(shí)現(xiàn)的消息隊(duì)列不適合大量消息堆積的場景,同時(shí)如果需要實(shí)現(xiàn)自動(dòng)分區(qū)消費(fèi)功能會(huì)比較復(fù)雜
- 適合消息量不是很大,且不是很復(fù)雜的場景
- 如果需要大量堆積消息和穩(wěn)定的多消費(fèi)者功能,可以使用自帶延遲消息的
RocketMQ
到此這篇關(guān)于Go+Redis實(shí)現(xiàn)延遲隊(duì)列實(shí)操的文章就介紹到這了,更多相關(guān)Go 延遲隊(duì)列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
利用ChatGPT編寫一個(gè)Golang圖像壓縮函數(shù)
這篇文章主要為大家詳細(xì)介紹了如何利用ChatGPT幫我們寫了一個(gè)Golang圖像壓縮函數(shù),文中的示例代碼簡潔易懂,感興趣的小伙伴可以嘗試一下2023-04-04詳解Go語言如何利用上下文進(jìn)行并發(fā)計(jì)算
在Go編程中,上下文(context)是一個(gè)非常重要的概念,它包含了與請求相關(guān)的信息,本文主要來和大家討論一下如何在并發(fā)計(jì)算中使用上下文,感興趣的可以了解下2024-02-02深入了解Go語言中web框架的中間件運(yùn)行機(jī)制
大家在使用iris框架搭建web系統(tǒng)時(shí),一定會(huì)用到中間件。那么你了解中間件的運(yùn)行機(jī)制嗎?你知道為什么在iris和gin框架的請求處理函數(shù)中要加c.Next()函數(shù)嗎?本文就和大家一起探究該問題的答案2023-02-02golang hack插件開發(fā)動(dòng)態(tài)鏈接庫實(shí)例探究
這篇文章主要為大家介紹了golang hack插件開發(fā)動(dòng)態(tài)鏈接庫實(shí)例探究,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2024-01-01