面试被问到Redis实现发布与订阅,手摸手教

发布时间:2025-05-14 04:29:34 作者:益华网络 来源:undefined 浏览量(0) 点赞(0)
摘要:简介 Redis发布与发布功能(Pub/Sub)是基于事件座位基本的通信机制,是目前应用比较普遍的通信模型,它的目的主要是解除消息的发布者与订阅者之间的耦合关系。 Redis作为消息发布和订阅之间的服务器,起到桥梁的作用,在Redis里面有一个channel的概念,也就是频道,发布者通过指定发布到某个

简介

Redis发布与发布功能(Pub/Sub)是基于事件座位基本的通信机制,是目前应用比较普遍的通信模型,它的目的主要是解除消息的发布者与订阅者之间的耦合关系。

Redis作为消息发布和订阅之间的服务器,起到桥梁的作用,在Redis里面有一个channel的概念,也就是频道,发布者通过指定发布到某个频道,然后只要有订阅者订阅了该频道,该消息就会发送给订阅者,原理图如下所示:

Redis同时也可以使用list类型实现消息队列(消息队列的实现以及应用场景会在下一篇文章继续讲解)。

Redis的发布与订阅的功能应用还是比较广泛的,它的应用场景有很多。比如:最常见的就是实现实时聊天的功能,还是有就是博客的粉丝文章的推送,当博主推送原创文章的时候,就会将文章实时推送给博主的粉丝。

简介完Redis的发布于订阅功能,下面就要来实操一下,包括linux命令的实操和java代码的实现。

命令实操

这里就假设各位读者都已经安装好自己的虚拟机环境和Redis了,若是没有安装好的,可以参考这一篇博文:https://www.cnblogs.com/zuidongfeng/p/8032505.html

我这里是已经安装好了Redis了,直接启动我们的Redis,我已经设置好了开机启动,上面的那篇博文有讲解怎么设置开机启动。

发布消息

Redis中发布消息的命令是publish,具体使用如下所示:

PUBLISH test "haha":test表示频道的名称,haha表示发布的内容,这样就完成了一个一个消息的发布,后面的返回(integer)0表示0人订阅。

订阅频道

于此同时再启动一个窗口,这个窗口作为订阅者,订阅者的命令subscribe,使用SUBSCRIBE test就表示订阅了test这个频道

订阅后返回的结果中由三条信息,第一个表示类型、第二个表示订阅的频道,第三个表示订阅的数量。接着在第一个窗口进行发布消息:

可以看到发布者发布的消息,订阅者都会实时的接收到,并发订阅者收到的信息中也会出现三条信息,分别表示:返回值的类型、频道名称、消息内容。

取消订阅

若是想取消之前的订阅可以使用unsubscribe命令,格式为:

unsubscribe  频道名称  // 取消之前订阅的test频道  unsubscribe  test 

输入命令后,返回以下结果: 

[root@pinyoyougou-docker src]# ./redis-cli   127.0.0.1:6379> UNSUBSCRIBE test  1) "unsubscribe" 2) "test"  3) (integer) 0 

它分别表示:返回值的类型、频道的名称、该频道订阅的数量。

按模式订阅

除了直接以特定的名城进行订阅,还可以按照模式进行订阅,模式的方式进行订阅可以一次订阅多个频道,按照模式进行订阅的命令为psubscribe,具体格式如下:

psubscribe  模式  // 表示订阅名称以ldc开头的频道  psubscribe  ldc* 

输入上面的命令后,返回如下结果: 

127.0.0.1:6379> PSUBSCRIBE ldc*  Reading messages... (press Ctrl-C to quit)  1) "psubscribe"  2) "ldc*"  3) (integer) 1 

这个也是非常简单,分别表示:返回的类型(表示按模式订阅类型)、订阅的模式、订阅数。

取消按模式订阅

假如你想取消之前的按模式订阅,可以使用punsubscribe来取消,具体格式: 

punsubscribe 模式  // 取消频道名称按照ldc开头的频道  punsubscribe ldc*

他的返回值,如下所示: 

127.0.0.1:6379> PUNSUBSCRIBE ldc*  1) "punsubscribe"  2) "ldc*"  3) (integer) 0 

这个就不多说了,表示的意思和上面的一样,可以看到上面的命令都是有规律的订阅SUBSCRIBE,取消就是UNSUBSCRIBE,前面加前缀UN,按模式订阅也是。

查看订阅消息

(1)你想查看某一个模式下订阅数是大于零的频道,可以使用如下格式的命令进行操作: 

pubsub channels 模式  // 查看频道名称以ldc模式开头的订阅数大于零的频道  pubsub channels ldc*

(2)假如你想查看某一个频道的订阅数,可以使用如下命令: 

pubsub numsub 频道名称 

(3)查看按照模式的订阅数,可以使用如下命令进行操作: 

pubsub numpat 

到这里以上的命令操作就基本结束了,下面就来代码实战。

代码实练

(1)首先第一步想要操作Redis,再SpringBoot项目中引入jedis的依赖,毕竟jedis是官方推荐使用操作Redis的工具。 

<dependency>      <groupId>redis.clients</groupId>      <artifactId>jedis</artifactId>      <version>2.9.0</version>  </dependency>

(2)然后创建发布者Publisher,用于消息的发布,具体代码如下: 

package com.ldc.org.myproject.demo.redis;  import java.io.BufferedReader;  import java.io.IOException;  import java.io.InputStreamReader;  import redis.clients.jedis.Jedis;  import redis.clients.jedis.JedisPool;  /**   * 发布者   * @author liduchang   *   */  public class Publisher extends Thread{   // 连接池    private final JedisPool jedisPool;   // 发布频道名称   private String name;   public Publisher(JedisPool jedisPool, String name) {    super();    this.jedisPool = jedisPool;    this.name = name;   }   @Override   public void run() {    // 获取要发布的消息    BufferedReader reader = new BufferedReader(new InputStreamReader(System.in));    // 获取连接    Jedis resource = jedisPool.getResource();    while (true) {     String message = null;     try {      message = reader.readLine();      if (!"exit".equals(message)) {       // 发布消息       resource.publish(name, "发布者:"+Thread.currentThread().getName()+"发布消息:"+message);      } else {       break;      }     } catch (IOException e) {      e.printStackTrace();     }    }  }  }

(3)接着创建订阅类Subscriber,并且继承JedisPubSub 类,重写onMessage、onSubscribe、onUnsubscribe三个方法,这三个方法的调用时机在注释上都有说明,具体的实现代码如下: 

package com.ldc.org.myproject.demo.redis;  import com.fasterxml.jackson.core.sym.Name;  import redis.clients.jedis.JedisPubSub;  /**   * 订阅者   * @author liduchang   */  public class Subscriber extends JedisPubSub {   //订阅频道名称   private String name;   public Subscriber(String name) {    this.name = name;   }   /**    * 订阅者收到消息时会调用    */   @Override   public void onMessage(String channel, String message) {    // TODO Auto-generated method stub    super.onMessage(channel, message);    System.out.println("频道:"+channel+"  接受的消息为:"+message);   }   /**    * 订阅了频道会被调用    */   @Override   public void onSubscribe(String channel, int subscribedChannels) {    System.out.println("订阅了频道:"+channel+"  订阅数为:"+subscribedChannels);   }   /**    * 取消订阅频道会被调用    */   @Override  public void onUnsubscribe(String channel, int subscribedChannels) {    System.out.println("取消订阅的频道:"+channel+"  订阅的频道数量为:"+subscribedChannels);   }  }

(4)这次创建的才是真正的订阅者SubThread,上面的Subscriber是指为了测试实订阅的时候或者发布消息,能够有信息输出: 

package com.ldc.org.myproject.demo.redis; import redis.clients.jedis.Jedis;  import redis.clients.jedis.JedisPool;  /**   * 订阅者线程   * @author liduchang   *   */  public class SubThread extends Thread {   private final JedisPool jedisPool;   private final Subscriber subscriber;   private String name;   public SubThread(JedisPool jedisPool,Subscriber subscriber,String name) {   super();    this.jedisPool = jedisPool;    this.subscriber = subscriber;    this.name = name;   }   @Override   public void run() {    Jedis jedis = null;    try {     jedis = jedisPool.getResource();     // 订阅频道为name     jedis.subscribe(subscriber, name);    } catch (Exception e) {     System.err.println("订阅失败");        e.printStackTrace();    } finally {     if (jedis!=null) {       // jedis.close();       //归还连接到redis池中      jedisPool.returnResource(jedis);     }    }   }  }

(5)后面就是测试了,分别测试发布与订阅的测试,发布者为TestPublisher,订阅者为TestSubscriber: 

package com.ldc.org.myproject.demo.redis;  import java.util.concurrent.ExecutorService;  import java.util.concurrent.Executors;  import java.util.concurrent.TimeUnit;  import redis.clients.jedis.JedisPool;  public class TestPublisher {   public static void main(String[] args) throws InterruptedException {    JedisPool jedisPool = new JedisPool("192.168.163.155");    // 向ldc频道发布消息    Publisher publisher = new Publisher(jedisPool, "ldc");    publisher.start();   }  }

订阅者 

package com.ldc.org.myproject.demo.redis;  import java.util.concurrent.ExecutorService;  import java.util.concurrent.Executors;  import java.util.concurrent.TimeUnit;  import redis.clients.jedis.JedisPool;  public class TestSubscriber1 {   public static void main(String[] args) throws InterruptedException {    JedisPool jedisPool = new JedisPool("192.168.163.155",6379);    Subscriber subscriber = new Subscriber("黎杜");    // 订阅ldc频道    SubThread thread= new SubThread(jedisPool, subscriber, "ldc");    thread.start();    Thread.sleep(600000);    // 取消订阅    subscriber.unsubscribe("ldc");   } }

这里为了测试方便就直接创建线程的方式,更好的话可以使用线程池的方式通过线程池的submit方法来执行线程,若是不用了可以使用shutdown方式关闭。

二维码

扫一扫,关注我们

声明:本文由【益华网络】编辑上传发布,转载此文章须经作者同意,并请附上出处【益华网络】及本页链接。如内容、图片有任何版权问题,请联系我们进行处理。

感兴趣吗?

欢迎联系我们,我们愿意为您解答任何有关网站疑难问题!

您身边的【网站建设专家】

搜索千万次不如咨询1次

主营项目:网站建设,手机网站,响应式网站,SEO优化,小程序开发,公众号系统,软件开发等

立即咨询 15368564009
在线客服
嘿,我来帮您!