文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

redis怎么实现队列阻塞、延时、发布和订阅

2023-07-02 08:42

关注

这篇“redis怎么实现队列阻塞、延时、发布和订阅”文章的知识点大部分人都不太理解,所以小编给大家总结了以下内容,内容详细,步骤清晰,具有一定的借鉴价值,希望大家阅读完这篇文章能有所收获,下面我们一起来看看这篇“redis怎么实现队列阻塞、延时、发布和订阅”文章吧。

Redis不仅可作为缓存服务器,还可以用作消息队列。它的列表类型天生支持用作消息队列。如下图所示:

redis怎么实现队列阻塞、延时、发布和订阅

由于Redis的列表是使用双向链表实现的,保存了头节点和尾节点,所以在列表的头部和尾部两边插入或获取元素都是非常快的,时间复杂度为O(1)。

普通队列

可以直接使用Redis的list数据类型实现消息队列,只需简单的两个指令lpush和rpop或者rpush和lpop。

下面使用redis的命令来模拟普通队列。
使用lpush命令生产消息:

>lpush queue:single 1"1">lpush queue:single 2"2">lpush queue:single 3"3"

使用rpop命令消费消息:

>rpop queue:single"1">rpop queue:single"2">rpop queue:single"3"

下面使用Java代码来实现普通队列。

生产者SingleProducer

package com.morris.redis.demo.queue.single;import redis.clients.jedis.Jedis;public class SingleProducer {    public static final String SINGLE_QUEUE_NAME = "queue:single";    public static void main(String[] args) {        Jedis jedis = new Jedis();        for (int i = 0; i < 100; i++) {            jedis.lpush(SINGLE_QUEUE_NAME, "hello " + i);        }        jedis.close();    }}

消费者SingleConsumer:

package com.morris.redis.demo.queue.single;import redis.clients.jedis.Jedis;import java.util.Objects;import java.util.concurrent.TimeUnit;public class SingleConsumer {    public static void main(String[] args) throws InterruptedException {        Jedis jedis = new Jedis();        while (true) {            String message = jedis.rpop(SingleProducer.SINGLE_QUEUE_NAME);            if(Objects.nonNull(message)) {                System.out.println(message);            } else {                TimeUnit.MILLISECONDS.sleep(500);            }        }    }}

上面的代码已经基本实现了普通队列的生产与消费,但是上述的例子中消息的消费者存在两个问题:

阻塞队列

消费者可以使用brpop指令从redis的list中获取数据,这个指令只有在有元素时才返回,没有则会阻塞直到超时返回null,于是消费端就不需要休眠后获取数据了,这样就相当于实现了一个阻塞队列,

使用redis的brpop命令来模拟阻塞队列。

>brpop queue:single 30

可以看到命令行阻塞在了brpop这里了,30s后没数据就返回。

Java代码实现如下:

生产者与普通队列的生产者一致。

消费者BlockConsumer:

package com.morris.redis.demo.queue.block;import redis.clients.jedis.Jedis;import java.util.List;public class BlockConsumer {    public static void main(String[] args) {        Jedis jedis = new Jedis();        while (true) {            // 超时时间为1s            List<String> messageList = jedis.brpop(1, BlockProducer.BLOCK_QUEUE_NAME);            if (null != messageList && !messageList.isEmpty()) {                System.out.println(messageList);            }        }    }}

缺点:无法实现一次生产多次消费。

发布订阅模式

Redis除了对消息队列提供支持外,还提供了一组命令用于支持发布/订阅模式。利用Redis的pub/sub模式可以实现一次生产多次消费的队列。

发布:PUBLISH指令可用于发布一条消息,格式:

PUBLISH channel message

返回值表示订阅了该消息的数量。

订阅:SUBSCRIBE指令用于接收一条消息,格式:

SUBSCRIBE channel

使用SUBSCRIBE指令后进入了订阅模式,但是不会接收到订阅之前publish发送的消息,这是因为只有在消息发出去前订阅才会接收到。在这个模式下其他指令,只能看到回复。

回复分为三种类型:

下面使用redis的命令来模拟发布订阅模式。

生产者:

127.0.0.1:6379> publish queue hello(integer) 1127.0.0.1:6379> publish queue hi(integer) 1

消费者:

127.0.0.1:6379> subscribe queueReading messages... (press Ctrl-C to quit)1) "subscribe"2) "queue"3) (integer) 11) "message"2) "queue"3) "hello"1) "message"2) "queue"3) "hi"

Java代码实现如下:

生产者PubsubProducer:

package com.morris.redis.demo.queue.pubsub;import redis.clients.jedis.Jedis;public class PubsubProducer {    public static final String PUBSUB_QUEUE_NAME = "queue:pubsub";    public static void main(String[] args) {        Jedis jedis = new Jedis();        for (int i = 0; i < 100; i++) {            jedis.publish(PUBSUB_QUEUE_NAME, "hello " + i);        }        jedis.close();    }}

消费者PubsubConsumer:

package com.morris.redis.demo.queue.pubsub;import redis.clients.jedis.Jedis;import redis.clients.jedis.JedisPubSub;public class PubsubConsumer {    public static void main(String[] args) throws InterruptedException {        Jedis jedis = new Jedis();        JedisPubSub jedisPubSub = new JedisPubSub() {            @Override            public void onMessage(String channel, String message) {                System.out.println("receive message: " + message);                if(message.indexOf("99") > -1) {                    this.unsubscribe();                }            }            @Override            public void onSubscribe(String channel, int subscribedChannels) {                System.out.println("subscribe channel: " + channel);            }            @Override            public void onUnsubscribe(String channel, int subscribedChannels) {                System.out.println("unsubscribe channel " + channel);            }        };        jedis.subscribe(jedisPubSub, PubsubProducer.PUBSUB_QUEUE_NAME);    }}

消费者可以启动多个,每个消费者都能收到所有的消息。

可以使用指令UNSUBSCRIBE退订,如果不加参数,则会退订所有由SUBSCRIBE指令订阅的频道。

Redis还支持基于通配符的消息订阅,使用指令PSUBSCRIBE (pattern subscribe),例如:

psubscribe channel.*

用PSUBSCRIBE指令订阅的频道也要使用指令PUNSUBSCRIBE指令退订,该指令无法退订SUBSCRIBE订阅的频道,同理UNSUBSCRIBE也不能退订PSUBSCRIBE指令订阅的频道。

同时PUNSUBSCRIBE指令通配符不会展开。例如:PUNSUBSCRIBE \*不会匹配到channel.\*,所以要取消订阅channel.\*就要这样写PUBSUBSCRIBE channel.\*。

Redis的pub/sub也有其缺点,那就是如果消费者下线,生产者的消息会丢失。

延时队列和优先级队列

Redis中有个数据类型叫Zset,其本质就是在数据类型Set的基础上加了个排序的功能而已,除了保存原始的数据value之外,还提供另一个属性score,这一属性在添加修改元素时候可以进行指定,每次指定后,Zset会自动重新按新的score值进行排序。

如果score字段设置为消息的优先级,优先级最高的消息排在第一位,这样就能实现一个优先级队列。

如果score字段代表的是消息想要执行时间的时间戳,将它插入Zset集合中,便会按照时间戳大小进行排序,也就是对执行时间先后进行排序,集合中最先要执行的消息就会排在第一位,这样的话,只需要起一个死循环线程不断获取集合中的第一个元素,如果当前时间戳大于等于该元素的score就将它取出来进行消费删除,就可以达到延时执行的目的,注意不需要遍历整个Zset集合,以免造成性能浪费。

下面使用redis的zset来模拟延时队列。

生产者:

127.0.0.1:6379> zadd queue:delay 1 order1 2 order2 3 order3(integer) 0

消费者:

127.0.0.1:6379> zrange queue:delay 0 0 withscores1) "order1"2) "1"127.0.0.1:6379> zrem queue:delay order1(integer) 1

Java代码如下:

生产者DelayProducer:

package com.morris.redis.demo.queue.delay;import redis.clients.jedis.Jedis;import java.util.Date;import java.util.Random;public class DelayProducer {    public static final String DELAY_QUEUE_NAME = "queue:delay";    public static void main(String[] args) {        Jedis jedis = new Jedis();        long now = new Date().getTime();        Random random = new Random();        for (int i = 0; i < 10; i++) {            int second = random.nextInt(30); // 随机订单失效时间            jedis.zadd(DELAY_QUEUE_NAME, now + second * 1000, "order"+i);        }        jedis.close();    }}

消费者:

package com.morris.redis.demo.queue.delay;import redis.clients.jedis.Jedis;import redis.clients.jedis.Tuple;import java.util.Date;import java.util.List;import java.util.Set;import java.util.concurrent.TimeUnit;public class DelayConsumer {    public static void main(String[] args) throws InterruptedException {        Jedis jedis = new Jedis();        while (true) {            long now = new Date().getTime();            Set<Tuple> tupleSet = jedis.zrangeWithScores(DelayProducer.DELAY_QUEUE_NAME, 0, 0);            if(tupleSet.isEmpty()) {                TimeUnit.MILLISECONDS.sleep(500);            } else {                for (Tuple tuple : tupleSet) {                    Double score = tuple.getScore();                    long time = score.longValue();                    if(time < now) {                        jedis.zrem(DelayProducer.DELAY_QUEUE_NAME, tuple.getElement());                        System.out.println("order[" + tuple.getElement() +"] is timeout at " + time);                    } else {                        TimeUnit.MILLISECONDS.sleep(500);                    }                    break;                }            }        }    }}

应用场景

延时队列可用于订单超时失效的场景
二级缓存(local+redis)中,当有缓存需要更新时,可以使用发布订阅模式通知其他服务器使得本地缓存失效。

以上就是关于“redis怎么实现队列阻塞、延时、发布和订阅”这篇文章的内容,相信大家都有了一定的了解,希望小编分享的内容对大家有帮助,若想了解更多相关的知识内容,请关注编程网行业资讯频道。

阅读原文内容投诉

免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     807人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     351人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     314人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     433人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     221人已做
    查看

相关文章

发现更多好内容

猜你喜欢

AI推送时光机
位置:首页-资讯-后端开发
咦!没有更多了?去看看其它编程学习网 内容吧
首页课程
资料下载
问答资讯