美文网首页Java
面试被问到Redis实现发布与订阅,我是这样回答的

面试被问到Redis实现发布与订阅,我是这样回答的

作者: 程序花生 | 来源:发表于2020-09-15 16:58 被阅读0次

    简介

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

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

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

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

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

    命令实操

    这里就假设各位读者都已经安装好自己的虚拟机环境和Redis了

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

    发布消息

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

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

    订阅频道

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

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

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

    取消订阅

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

    unsubscribe  频道名称

    // 取消之前订阅的test频道

    unsubscribetest

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

    [root@pinyoyougou-dockersrc]#./redis-cli

    127.0.0.1:6379>UNSUBSCRIBEtest

    1)"unsubscribe"

    2)"test"

    3)(integer)0

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

    按模式订阅

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

    psubscribe  模式

    // 表示订阅名称以ldc开头的频道 

    psubscribe  ldc*

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

    127.0.0.1:6379>PSUBSCRIBEldc*

    Readingmessages...(pressCtrl-Ctoquit)

    1)"psubscribe"

    2)"ldc*"

    3)(integer)1

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

    取消按模式订阅

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

    punsubscribe 模式

    // 取消频道名称按照ldc开头的频道 

    punsubscribe ldc*

    它的返回值,如下所示:

    127.0.0.1:6379>PUNSUBSCRIBEldc*

    1)"punsubscribe"

    2)"ldc*"

    3)(integer)0

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

    查看订阅消息

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

    pubsub channels 模式

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

    pubsub channels ldc*

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

    pubsubnumsub 频道名称

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

    pubsubnumpat

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

    代码实练

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

    redis.clients

    jedis

    2.9.0

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

    packagecom.ldc.org.myproject.demo.redis;

    importjava.io.BufferedReader;

    importjava.io.IOException;

    importjava.io.InputStreamReader;

    importredis.clients.jedis.Jedis;

    importredis.clients.jedis.JedisPool;

    /** 

    * 发布者 

    *@authorliduchang

    */

    publicclassPublisherextendsThread{

    // 连接池 

    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三个方法,这三个方法的调用时机在注释上都有说明,具体的实现代码如下:

    packagecom.ldc.org.myproject.demo.redis;

    importcom.fasterxml.jackson.core.sym.Name;

    importredis.clients.jedis.JedisPubSub;

    /** 

    * 订阅者 

    *@authorliduchang

    */

    publicclassSubscriberextendsJedisPubSub{

    //订阅频道名称 

    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是指为了测试实订阅的时候或者发布消息,能够有信息输出:

    packagecom.ldc.org.myproject.demo.redis;

    importredis.clients.jedis.Jedis;

    importredis.clients.jedis.JedisPool;

    /** 

    * 订阅者线程 

    *@authorliduchang

    */

    publicclassSubThreadextendsThread{

    privatefinalJedisPool jedisPool;

    privatefinalSubscriber subscriber;

    privateString name;

    publicSubThread(JedisPool jedisPool,Subscriber subscriber,String name){

    super();

    this.jedisPool = jedisPool;

    this.subscriber = subscriber;

    this.name = name;

    }@Override

    publicvoidrun(){

    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:

    packagecom.ldc.org.myproject.demo.redis;

    importjava.util.concurrent.ExecutorService;

    importjava.util.concurrent.Executors;

    importjava.util.concurrent.TimeUnit;

    importredis.clients.jedis.JedisPool;

    publicclassTestPublisher{

    publicstaticvoidmain(String[] args)throwsInterruptedException{

    JedisPool jedisPool =newJedisPool("192.168.163.155");

    // 向ldc频道发布消息 

      Publisher publisher = new Publisher(jedisPool, "ldc"); 

      publisher.start(); 

    }

    订阅者

    packagecom.ldc.org.myproject.demo.redis;

    importjava.util.concurrent.ExecutorService;

    importjava.util.concurrent.Executors;

    importjava.util.concurrent.TimeUnit;

    importredis.clients.jedis.JedisPool;

    publicclassTestSubscriber1{

    publicstaticvoidmain(String[] args)throwsInterruptedException{

    JedisPool jedisPool =newJedisPool("192.168.163.155",6379);

    Subscriber subscriber =newSubscriber("黎杜");

    // 订阅ldc频道 

      SubThread thread= new SubThread(jedisPool, subscriber, "ldc"); 

      thread.start(); 

      Thread.sleep(600000); 

      // 取消订阅 

      subscriber.unsubscribe("ldc"); 

    }

    }

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

    作者:非科班的科班

    链接:https://developer.51cto.com/art/202009/626162.htm#topx

    相关文章

      网友评论

        本文标题:面试被问到Redis实现发布与订阅,我是这样回答的

        本文链接:https://www.haomeiwen.com/subject/nmmfyktx.html