面试被问到Redis实现发布与订阅,手摸手教
简介
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
方式关闭。
好了这一期的Redis的实现订阅与发布的讲解就说完了,我们下一期在讲解Redis的集群的知识,下期再见。
【热门阅读】