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

Posted python爬虫人工智能大数据

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了面试被问到Redis实现发布与订阅，手摸手教相关的知识，希望对你有一定的参考价值。

简介

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

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


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

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

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

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

命令实操

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

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

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

发布消息

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

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

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

订阅频道

于此同时再启动一个窗口，这个窗口作为订阅者，订阅者的命令subscribe，使用SUBSCRIBE test就表示订阅了test这个频道面试被问到Redis实现发布与订阅，手摸手教订阅后返回的结果中由三条信息，第一个表示类型、第二个表示订阅的频道，第三个表示订阅的数量。接着在第一个窗口进行发布消息：

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

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

取消订阅

若是想取消之前的订阅可以使用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方式关闭。

好了这一期的Redis的实现订阅与发布的讲解就说完了，我们下一期在讲解Redis的集群的知识，下期再见。


【热门阅读】

[1] 
[2] 
[3] 
[4] 
 [5] 

 [6] 



以上是关于面试被问到Redis实现发布与订阅，手摸手教的主要内容，如果未能解决你的问题，请参考以下文章

iOS动画进阶 - 手摸手教你写ShineButton动画

手摸手教你微信小程序开发之自定义组件

手摸手教你 HTTPS 的原理（通俗易懂！）

手摸手教你玩转 vue render 函数

张三李四手摸手教你 HTTPS 的原理（通俗易懂！）

手摸手教你封装几个Vue3中很有用的组合式API