上一篇基于 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 body) msg.delaytime="tm.Unix()" senddata, _ p.rediscmd.zadd(topicname+zsetsuffix, redis.z{score: float64(tm.unix()), member: string(senddata)}).err() < code></=>
使用,比如我们想过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 { 0, currenttime) pip.zremrangebyscore(topicname, "0", strconv.formatint(currenttime, 10)) nil }) if !="nil" log.printf("zset pip error: err) continue } rev for revbody msg json.unmarshal([]byte(revbody.member.(string)), msg) s.handler s.handler.handlemessage(msg) }() < code></-s.ctx.done():>
Original: https://www.cnblogs.com/li-peng/p/12697110.html
Author: li-peng
Title: 玩转redis-延时消息队列
原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/529058/
转载文章受原作者版权保护。转载请注明原作者出处！