欧美bbbwbbbw肥妇,免费乱码人妻系列日韩,一级黄片

Go+Redis實現(xiàn)延遲隊列實操

 更新時間:2022年09月14日 08:54:28   作者:jiaxwu???????  
這篇文章主要介紹了Go+Redis實現(xiàn)延遲隊列實操,延遲隊列是一種非常使用的數(shù)據(jù)結(jié)構(gòu),我們經(jīng)常有需要延遲推送處理消息的場景,比如延遲60秒發(fā)送短信,延遲30分鐘關(guā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ù)

    這篇文章主要為大家詳細介紹了如何利用ChatGPT幫我們寫了一個Golang圖像壓縮函數(shù),文中的示例代碼簡潔易懂,感興趣的小伙伴可以嘗試一下
    2023-04-04
  • Go語言新寵:pdqsort排序算法的完美打造

    Go語言新寵:pdqsort排序算法的完美打造

    pdqsort是一種新的排序算法,特別適用于Go語言,它是由Go語言團隊開發(fā)的,旨在提供高效且穩(wěn)定的排序算法,pdqsort采用了一種分治的策略,將數(shù)組分成小塊進行排序,然后再合并這些塊,需要的朋友可以參考下
    2023-10-10
  • Golang字符串的拼接方法匯總

    Golang字符串的拼接方法匯總

    字符串拼接在日常開發(fā)中是很常見的需求,今天我們來探討下如何用golang來實現(xiàn)字符串的拼接
    2018-10-10
  • golang判斷key是否在map中的代碼

    golang判斷key是否在map中的代碼

    這篇文章主要介紹了golang判斷key是否在map中的代碼,具有很好的參考價值,希望對大家有所幫助。一起跟隨小編過來看看吧
    2021-04-04
  • 淺談Go語言中字符串和數(shù)組

    淺談Go語言中字符串和數(shù)組

    這篇文章主要簡單介紹了Go語言中字符串和數(shù)組的使用方法和申明方式,需要的朋友可以參考下
    2015-01-01
  • 詳解Go語言如何利用上下文進行并發(fā)計算

    詳解Go語言如何利用上下文進行并發(fā)計算

    在Go編程中,上下文(context)是一個非常重要的概念,它包含了與請求相關(guān)的信息,本文主要來和大家討論一下如何在并發(fā)計算中使用上下文,感興趣的可以了解下
    2024-02-02
  • 深入了解Go語言中web框架的中間件運行機制

    深入了解Go語言中web框架的中間件運行機制

    大家在使用iris框架搭建web系統(tǒng)時,一定會用到中間件。那么你了解中間件的運行機制嗎?你知道為什么在iris和gin框架的請求處理函數(shù)中要加c.Next()函數(shù)嗎?本文就和大家一起探究該問題的答案
    2023-02-02
  • Go 代碼生成工具詳解

    Go 代碼生成工具詳解

    這篇文章主要介紹了Go 代碼生成工具詳解,本文給大家介紹的非常詳細,對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下
    2021-03-03
  • golang hack插件開發(fā)動態(tài)鏈接庫實例探究

    golang hack插件開發(fā)動態(tài)鏈接庫實例探究

    這篇文章主要為大家介紹了golang hack插件開發(fā)動態(tài)鏈接庫實例探究,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪
    2024-01-01
  • Golang環(huán)境搭建及打包和工具鏈詳解

    Golang環(huán)境搭建及打包和工具鏈詳解

    這篇文章主要介紹了Golang環(huán)境搭建及打包和工具鏈,本文給大家介紹的非常詳細,對大家的學(xué)習(xí)或工作具有一定的參考借鑒價值,需要的朋友可以參考下
    2023-04-04

最新評論