玩转redis-延时消息队列-阿里云开发者社区

开发者社区> lpxxn> 正文

玩转redis-延时消息队列

简介:
+关注继续查看

上一篇基于redis的list实现了一个简单的消息队列:玩转redis-简单消息队列

源码地址 使用demo

产品经理经常说的一句话,我们不光要有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

延时队列的实现思路

总体的思路很简单,就是每一个valuescore保存的是时间,也就是说,在添加一个元素时他的score是当前时间+延时的时间。轮循获取数据时,查找小于或等于当前时间的数据项,就是具体的延时消息。

还有一个问题,就是ZRANGEBYSCORElistpop不同,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)
                    }
                }
            }
        }
    }()
}

版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
阿里云服务器怎么设置密码?怎么停机?怎么重启服务器?
如果在创建实例时没有设置密码,或者密码丢失,您可以在控制台上重新设置实例的登录密码。本文仅描述如何在 ECS 管理控制台上修改实例登录密码。
9975 0
当 Redis 发生高延迟时,到底发生了什么
Redis 是一种内存数据库,将数据保存在内存中,读写效率要比传统的将数据保存在磁盘上的数据库要快很多。但是 Redis 也会发生延迟时,这是就需要我们对其产生原因有深刻的了解,以便于快速排查问题,解决 Redis的延迟问题
1603 0
阿里云服务器如何登录?阿里云服务器的三种登录方法
购买阿里云ECS云服务器后如何登录?场景不同,阿里云优惠总结大概有三种登录方式: 登录到ECS云服务器控制台 在ECS云服务器控制台用户可以更改密码、更换系.
13738 0
大消息大小配额。若要增加配额,请使用相应绑定元素上的 MaxReceivedMessageSize
.net引用webservice 发生错误 应该是因为获得的数据大于webconfig设定的大小   解决: 在web。config中修改 MaxReceivedMessageSize 的大小 并保持与maxBufferSize 一致
583 0
怎样使用阿里云消息服务?
阿里云消息服务(Message Service)是一种高效、可靠、安全、便捷、可弹性扩展的分布式消息服务。MNS能够帮助应用开发者在他们应用的分布式组件上自由的传递数据、通知消息,构建松耦合系统。 消息服务同时支持各种类型消息推送,其中和短信前后端的无缝整合更高效的为用户提供了大批量短信发送能力。
1717 0
+关注
lpxxn
山外青山,楼外楼。
174
文章
0
问答
文章排行榜
最热
最新
相关电子书
更多
《2021云上架构与运维峰会演讲合集》
立即下载
《零基础CSS入门教程》
立即下载
《零基础HTML入门教程》
立即下载