玩轉redis-延時消息隊列
上一篇基於redis的list實現了一個簡單的消息隊列：玩轉redis-簡單消息隊列
產品經理經常說的一句話，我們不光要有X功能，還要Y功能，這樣客戶才能更滿意。同樣的，只有簡單消息隊列是不夠的，還要有延時消息隊列才能算是一個完整的消息隊列。
看看redis的命令，放眼望去，的有序集合(sorted set)就是一個很好用的命令，完全可以用他做一個延時消息隊列
redis有序集合(sorted set)
redis有序集合，每個元素都會關聯一個double類型的分數。redis正是通過分數來為集合中的成員進行從小到大的排序。
有序集合的成員是唯一的,但分數(score)卻可以重複。
簡單操作
添加數據
127.0.0.1:6379> ZADD testSet1 5 a
(integer) 1
127.0.0.1:6379> ZADD testSet1 1 b 8 c 7 d
(integer) 3
讀取
127.0.0.1:6379> ZRANGEBYSCORE testSet1 0 3
1) "b"
127.0.0.1:6379> ZRANGEBYSCORE testSet1 0 5
1) "b"
2) "a"
也可以把score打出來
127.0.0.1:6379> ZRANGEBYSCORE testSet1 -inf 5 WITHSCORES
1) "b"
2) "1"
3) "a"
4) "5"
查出所有的數據
127.0.0.1:6379> ZRANGEBYSCORE testSet1 -inf inf
1) "b"
2) "a"
3) "d"
4) "c"
刪除數據
ZREMRANGEBYSCORE testSet1 0 2
延時隊列的實現思路
總體的思路很簡單，就是每一個value的score保存的是時間，也就是說，在添加一個元素時他的score是當前時間+延時的時間。輪循獲取數據時，查找小於或等於當前時間的數據項，就是具體的延時消息。
還有一個問題，就是
ZRANGEBYSCORE和list的pop不同，pop是取出元素並且會把元素在list中刪除。ZRANGEBYSCORE只會取出數據不會把數據從sorted set中刪除。解決方法1，利用redis的事務，先ZRANGEBYSCORE取出數據，然後再用ZREMRANGEBYSCORE把數據刪除。
具體實現-code
添加延時消息，參數delay就是我們要延時多久:
func (p *Producer) PublishDelayMsg(topicName string, body []byte, delay time.Duration) error {
if delay <= 0 {
return errors.New("delay need great than zero")
}
tm := time.Now().Add(delay)
msg := NewMessage("", body)
msg.DelayTime = tm.Unix()
sendData, _ := json.Marshal(msg)
return p.redisCmd.ZAdd(topicName+zsetSuffix, redis.Z{Score: float64(tm.Unix()), Member: string(sendData)}).Err()
}
使用,比如我們想過1秒再處理
producer.PublishDelayMsg(topicName, body, time.Second)
讀取消息並處理
這就比較簡單了，就是在一個ticker里循環讀取小於或等於當前時間的數據：
func (s *consumer) startGetDelayMessage() {
go func() {
ticker := time.NewTicker(s.options.RateLimitPeriod)
defer func() {
log.Println("stop get delay message.")
ticker.Stop()
}()
topicName := s.topicName + zsetSuffix
for {
currentTime := time.Now().Unix()
select {
case <-s.ctx.Done():
log.Printf("context Done msg: %#v \n", s.ctx.Err())
return
case <-ticker.C:
var valuesCmd *redis.ZSliceCmd
_, err := s.redisCmd.TxPipelined(func(pip redis.Pipeliner) error {
valuesCmd = pip.ZRangeWithScores(topicName, 0, currentTime)
pip.ZRemRangeByScore(topicName, "0", strconv.FormatInt(currentTime, 10))
return nil
})
if err != nil {
log.Printf("zset pip error: %#v \n", err)
continue
}
rev := valuesCmd.Val()
for _, revBody := range rev {
msg := &Message{}
json.Unmarshal([]byte(revBody.Member.(string)), msg)
if s.handler != nil {
s.handler.HandleMessage(msg)
}
}
}
}
}()
}
