首页-教程-服务器/数据库

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

 2732

本篇文章给大家带来了关于Redis的相关知识，其中主要介绍了关于怎么实现队列的阻塞、延时、发布和订阅的相关问题，下面一起来看一下，希望对大家有帮助。


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


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

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

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

普通队列

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

lpush+rpop：左进右出的队列

rpush+lpop：左出右进的队列

下面使用redis的命令来模拟普通队列。

使用lpush命令生产消息：

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

使用rpop命令消费消息：

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

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

生产者SingleProducer

  1. package com.morris.redis.demo.queue.single;
  2. import redis.clients.jedis.Jedis;
  3. /**
  4.  * 生产者
  5.  */
  6. public class SingleProducer {
  7.  
  8.     public static final String SINGLE_QUEUE_NAME = "queue:single";
  9.  
  10.     public static void main(String[] args) {
  11.         Jedis jedis = new Jedis();
  12.         for (int i = 0; i < 100; i++) {
  13.             jedis.lpush(SINGLE_QUEUE_NAME, "hello " + i);
  14.         }
  15.         jedis.close();
  16.     }
  17. }

消费者SingleConsumer：

  1. package com.morris.redis.demo.queue.single;
  2. import redis.clients.jedis.Jedis;
  3. import java.util.Objects;
  4. import java.util.concurrent.TimeUnit;
  5. /**
  6.  * 消费者
  7.  */
  8. public class SingleConsumer {
  9.  
  10.     public static void main(String[] args) throws InterruptedException {
  11.         Jedis jedis = new Jedis();
  12.         while (true) {
  13.             String message = jedis.rpop(SingleProducer.SINGLE_QUEUE_NAME);
  14.             if(Objects.nonNull(message)) {
  15.                 System.out.println(message);
  16.             } else {
  17.                 TimeUnit.MILLISECONDS.sleep(500);
  18.             }
  19.         }
  20.     }
  21. }

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

消费者需要不停的调用rpop方法查看redis的list中是否有待处理的数据（消息）。每调用一次都会发起一次连接，有可能list中没有数据，造成大量的空轮询，导致造成不必要的浪费。也许你可以使用Thread.sleep()等方法让消费者线程隔一段时间再消费，如果睡眠时间过长，这样不能处理一些时效性要求高的消息，睡眠时间过短，也会在连接上造成比较大的开销。

如果生产者速度大于消费者消费速度，消息队列长度会一直增大，时间久了会占用大量内存空间。


阻塞队列

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

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

  1. >brpop queue:single 30

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

Java代码实现如下：

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

消费者BlockConsumer：

  1. package com.morris.redis.demo.queue.block;
  2. import redis.clients.jedis.Jedis;
  3. import java.util.List;
  4. /**
  5.  * 消费者
  6.  */
  7. public class BlockConsumer {
  8.  
  9.     public static void main(String[] args) {
  10.         Jedis jedis = new Jedis();
  11.         while (true) {
  12.             // 超时时间为1s
  13.             List<String> messageList = jedis.brpop(1, BlockProducer.BLOCK_QUEUE_NAME);
  14.             if (null != messageList && !messageList.isEmpty()) {
  15.                 System.out.println(messageList);
  16.             }
  17.         }
  18.     }
  19. }

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


发布订阅模式

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

发布：PUBLISH指令可用于发布一条消息，格式：

  1. PUBLISH channel message

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

订阅：SUBSCRIBE指令用于接收一条消息，格式：

  1. SUBSCRIBE channel

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

回复分为三种类型：

如果为subscribe，第二个值表示订阅的频道，第三个值表示是已订阅的频道的数量

如果为message(消息)，第二个值为产生该消息的频道，第三个值为消息

如果为unsubscribe，第二个值表示取消订阅的频道，第三个值表示当前客户端的订阅数量。

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

生产者：

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

消费者：

  1. 127.0.0.1:6379> subscribe queue
  2. Reading 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：

  1. package com.morris.redis.demo.queue.pubsub;
  2. import redis.clients.jedis.Jedis;
  3. /**
  4.  * 生产者
  5.  */
  6. public class PubsubProducer {
  7.  
  8.     public static final String PUBSUB_QUEUE_NAME = "queue:pubsub";
  9.  
  10.     public static void main(String[] args) {
  11.         Jedis jedis = new Jedis();
  12.         for (int i = 0; i < 100; i++) {
  13.             jedis.publish(PUBSUB_QUEUE_NAME, "hello " + i);
  14.         }
  15.         jedis.close();
  16.     }
  17. }

消费者PubsubConsumer：

  1. package com.morris.redis.demo.queue.pubsub;
  2. import redis.clients.jedis.Jedis;
  3. import redis.clients.jedis.JedisPubSub;
  4. /**
  5.  * 消费者
  6.  */
  7. public class PubsubConsumer {
  8.  
  9.     public static void main(String[] args) throws InterruptedException {
  10.         Jedis jedis = new Jedis();
  11.  
  12.         JedisPubSub jedisPubSub = new JedisPubSub() {
  13.  
  14.             @Override
  15.             public void onMessage(String channel, String message) {
  16.                 System.out.println("receive message: " + message);
  17.                 if(message.indexOf("99") > -1) {
  18.                     this.unsubscribe();
  19.                 }
  20.             }
  21.  
  22.             @Override
  23.             public void onSubscribe(String channel, int subscribedChannels) {
  24.                 System.out.println("subscribe channel: " + channel);
  25.             }
  26.  
  27.             @Override
  28.             public void onUnsubscribe(String channel, int subscribedChannels) {
  29.                 System.out.println("unsubscribe channel " + channel);
  30.             }
  31.         };
  32.  
  33.         jedis.subscribe(jedisPubSub, PubsubProducer.PUBSUB_QUEUE_NAME);
  34.     }
  35. }

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

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

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

  1. 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来模拟延时队列。

生产者：

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

消费者：

  1. 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：

  1. package com.morris.redis.demo.queue.delay;
  2. import redis.clients.jedis.Jedis;
  3. import java.util.Date;import java.util.Random;
  4. /**
  5.  * 生产者
  6.  */
  7. public class DelayProducer {
  8.  
  9.     public static final String DELAY_QUEUE_NAME = "queue:delay";
  10.  
  11.     public static void main(String[] args) {
  12.         Jedis jedis = new Jedis();
  13.         long now = new Date().getTime();
  14.         Random random = new Random();
  15.         for (int i = 0; i < 10; i++) {
  16.             int second = random.nextInt(30); // 随机订单失效时间
  17.             jedis.zadd(DELAY_QUEUE_NAME, now + second * 1000, "order"+i);
  18.         }
  19.         jedis.close();
  20.     }
  21. }

消费者：

  1. package com.morris.redis.demo.queue.delay;
  2. import redis.clients.jedis.Jedis;
  3. import redis.clients.jedis.Tuple;
  4. import java.util.Date;import java.util.List;
  5. import java.util.Set;import java.util.concurrent.TimeUnit;
  6. /**
  7.  * 消费者
  8.  */
  9. public class DelayConsumer {
  10.  
  11.     public static void main(String[] args) throws InterruptedException {
  12.         Jedis jedis = new Jedis();
  13.         while (true) {
  14.             long now = new Date().getTime();
  15.             Set<Tuple> tupleSet = jedis.zrangeWithScores(DelayProducer.DELAY_QUEUE_NAME, 0, 0);
  16.             if(tupleSet.isEmpty()) {
  17.                 TimeUnit.MILLISECONDS.sleep(500);
  18.             } else {
  19.                 for (Tuple tuple : tupleSet) {
  20.                     Double score = tuple.getScore();
  21.                     long time = score.longValue();
  22.                     if(time < now) {
  23.                         jedis.zrem(DelayProducer.DELAY_QUEUE_NAME, tuple.getElement());
  24.                         System.out.println("order[" + tuple.getElement() +"] is timeout at " + time);
  25.                     } else {
  26.                         TimeUnit.MILLISECONDS.sleep(500);
  27.                     }
  28.                     break;
  29.                 }
  30.             }
  31.         }
  32.     }
  33. }


应用场景

延时队列可用于订单超时失效的场景

二级缓存（local+redis）中，当有缓存需要更新时，可以使用发布订阅模式通知其他服务器使得本地缓存失效。


TAG标签：
本文网址：https://www.zztuku.com/index.php/detail-12078.html
站长图库 - Redis怎么实现队列的阻塞、延时、发布和订阅
申明：本文转载于《CSDN》，如有侵犯，请 联系我们 删除。

评论（0）条

您还没有登录，请 登录 后发表评论！

提示：请勿发布广告垃圾评论，否则封号处理！！

    编辑推荐