Go+Redis實現(xiàn)延遲隊列實操
前言
延遲隊列是一種非常使用的數(shù)據(jù)結(jié)構(gòu),我們經(jīng)常有需要延遲推送處理消息的場景,比如延遲60秒發(fā)送短信,延遲30分鐘關(guān)閉訂單,消息消費失敗延遲重試等等。
一般我們實現(xiàn)延遲消息都需要依賴底層的有序結(jié)構(gòu),比如堆,而Redis剛好提供了zset
這種數(shù)據(jù)類型,它的底層實現(xiàn)是哈希表+跳表
,也是一種有序的結(jié)構(gòu),所以這篇文章主要是使用Go+Redis來實現(xiàn)延遲隊列。
當(dāng)然Redis本身并不支持延遲隊列,所以我們只是實現(xiàn)一個比較簡單的延遲隊列,而且Redis不太適合大量消息堆積,所以只適合比較簡單的場景,如果需要更加強大穩(wěn)定的消息隊列,可以使用RocketMQ
等自帶延遲消息的消息隊列。
我們這里先定一下我們要實現(xiàn)的幾個目標:
- 消息必須至少被消費一次
- 多個生產(chǎn)者
- 多個消費者
然后我們定義一個簡單的接口:
- Push(msg) error:添加消息到隊列
- Consume(topic, batchSize, func(msg) error):消費消息
簡單的實現(xiàn)
- 每個主題最多可以被一個消費者消費,因為不會對主題進行分區(qū)
- 但是可以多個生產(chǎn)者同時進行生產(chǎn),因為Push操作是原子的
- 同時需要消費操作返回值error為nil才刪除消息,保證消息至少被消費一次
定義消息
這個消息參考了Kafka的消息結(jié)構(gòu):
- Topic可以是某個隊列的名字
- Key是消息的唯一標識,在一個隊列里面不可以重復(fù)
- Body是消息的內(nèi)容
- Delay是消息的延遲時間
- ReadyTime是消息準備好執(zhí)行的時間
// Msg 消息 type Msg struct { Topic string // 消息的主題 Key string // 消息的Key Body []byte // 消息的Body Delay time.Duration // 延遲時間(秒) ReadyTime time.Time // 消息準備好執(zhí)行的時間(now + delay) }
Push
由于我們需要把消息的Body存儲到Hash
,把消息的ReadyTime存儲到ZSet
,所以我們需要一個簡單的Lua腳本來保證這兩個操作是原子的
。
同時我們不會覆蓋已經(jīng)存在的相同Key的消息。
const delayQueuePushRedisScript = ` -- KEYS[1]: topicZSet -- KEYS[2]: topicHash -- ARGV[1]: 消息的Key -- ARGV[2]: 消息的Body -- ARGV[3]: 消息準備好執(zhí)行的時間 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
其中第二個參數(shù)batchSize
表示用于批量獲取已經(jīng)準備好執(zhí)行的消息,減少網(wǎng)絡(luò)請求。
fn
是對消息進行處理的函數(shù),它有一個返回值error
,如果是nil
才表示消息消費成功,然后調(diào)用刪除腳本把成功消費的消息給刪除(需要原子的刪除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í)行的消息 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() // 如果獲取出錯或者獲取不到消息,則休眠一秒 if err != nil || len(zs) == 0 { time.Sleep(time.Second) continue } // 遍歷每個消息 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) } } }
存在的問題
如果多個線程同時調(diào)用Consume
函數(shù),那么多個線程會拉取相同的可執(zhí)行的消息,造成消息重復(fù)的被消費。
多消費者實現(xiàn)
- 每個主題最多可以被分區(qū)個數(shù)個消費者消費,會對主題進行分區(qū)
定義消息
- 我們添加了一個Partition字段表示消息的分區(qū)號
// Msg 消息 type Msg struct { Topic string // 消息的主題 Key string // 消息的Key Body []byte // 消息的Body Partition int // 分區(qū)號 Delay time.Duration // 延遲時間(秒) ReadyTime time.Time // 消息準備好執(zhí)行的時間 }
Push
代碼與SimpleRedisDelayQueue的Push相同,只是我們會使用Msg里面的Partition字段對主題進行分區(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多加了一個partition
參數(shù)用于指定消費的分區(qū)。
func (q *PartitionRedisDelayQueue) Consume(topic string, batchSize, partition int, fn func(msg *Msg) error) { // ... }
存在的問題
一個比較大的問題就是我們需要手動指定分區(qū)而不是自動分配分區(qū),這個問題對于Push操作解決起來比較容易,可以通過哈希算法對Key進行哈希取模進行分區(qū),比如murmur3。但是對于Consume就比較復(fù)雜,因為我們必須記錄哪個分區(qū)已經(jīng)被消費者消費了。如果真的需要更加復(fù)雜的場景還是建議使用RocketMQ
、Kafka
等消息隊列進行實現(xiàn)。
總結(jié)
- 使用Redis的ZSet可以很容易的實現(xiàn)一個高性能消息隊列
- 但是Redis的ZSet實現(xiàn)的消息隊列不適合大量消息堆積的場景,同時如果需要實現(xiàn)自動分區(qū)消費功能會比較復(fù)雜
- 適合消息量不是很大,且不是很復(fù)雜的場景
- 如果需要大量堆積消息和穩(wěn)定的多消費者功能,可以使用自帶延遲消息的
RocketMQ
到此這篇關(guān)于Go+Redis實現(xiàn)延遲隊列實操的文章就介紹到這了,更多相關(guān)Go 延遲隊列內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
利用ChatGPT編寫一個Golang圖像壓縮函數(shù)
這篇文章主要為大家詳細介紹了如何利用ChatGPT幫我們寫了一個Golang圖像壓縮函數(shù),文中的示例代碼簡潔易懂,感興趣的小伙伴可以嘗試一下2023-04-04golang hack插件開發(fā)動態(tài)鏈接庫實例探究
這篇文章主要為大家介紹了golang hack插件開發(fā)動態(tài)鏈接庫實例探究,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪2024-01-01