首页 资讯频道 互联频道 智能频道 网络 数据频道 安全频道 服务器频道 存储频道

Redis实现发布与订阅 命令的实际操作步骤

2020-09-16 08:32:14 来源 : 数据库开发公众号

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

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

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

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

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

命令实操

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

发布消息

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

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

订阅频道

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

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

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

取消订阅

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

unsubscribe频道名称

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

unsubscribetest

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

[[email protected]]#./redis-cli

127.0.0.1:6379>UNSUBSCRIBEtest

1)"unsubscribe"

2)"test"

3)(integer)0

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

按模式订阅

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

psubscribe模式

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

psubscribeldc*

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

127.0.0.1:6379>PSUBSCRIBEldc*

Readingmessages...(pressCtrl-Ctoquit)

1)"psubscribe"

2)"ldc*"

3)(integer)1

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

取消按模式订阅

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

punsubscribe模式

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

punsubscribeldc*

他的返回值,如下所示:

127.0.0.1:6379>PUNSUBSCRIBEldc*

1)"punsubscribe"

2)"ldc*"

3)(integer)0

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

查看订阅消息

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

pubsubchannels模式

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

pubsubchannelsldc*

(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{

//连接池

privatefinalJedisPooljedisPool;

//发布频道名称

privateStringname;

publicPublisher(JedisPooljedisPool,Stringname){

super();

this.jedisPool=jedisPool;

this.name=name;

}

@Override

publicvoidrun(){

//获取要发布的消息

BufferedReaderreader=newBufferedReader(newInputStreamReader(System.in));

//获取连接

Jedisresource=jedisPool.getResource();

while(true){

Stringmessage=null;

try{

message=reader.readLine();

if(!"exit".equals(message)){

//发布消息

resource.publish(name,"发布者:"+Thread.currentThread().getName()+"发布消息:"+message);

}else{

break;

}

}catch(IOExceptione){

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{

//订阅频道名称

privateStringname;

publicSubscriber(Stringname){

this.name=name;

}

/**

*订阅者收到消息时会调用

*/

@Override

publicvoidonMessage(Stringchannel,Stringmessage){

//TODOAuto-generatedmethodstub

super.onMessage(channel,message);

System.out.println("频道:"+channel+"接受的消息为:"+message);

}

/**

*订阅了频道会被调用

*/

@Override

publicvoidonSubscribe(Stringchannel,intsubscribedChannels){

System.out.println("订阅了频道:"+channel+"订阅数为:"+subscribedChannels);

}

/**

*取消订阅频道会被调用

*/

@Override

publicvoidonUnsubscribe(Stringchannel,intsubscribedChannels){

System.out.println("取消订阅的频道:"+channel+"订阅的频道数量为:"+subscribedChannels);

}

}

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

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

importredis.clients.jedis.Jedis;

importredis.clients.jedis.JedisPool;

/**

*订阅者线程

*@authorliduchang

*

*/

publicclassSubThreadextendsThread{

privatefinalJedisPooljedisPool;

privatefinalSubscribersubscriber;

privateStringname;

publicSubThread(JedisPooljedisPool,Subscribersubscriber,Stringname){

super();

this.jedisPool=jedisPool;

this.subscriber=subscriber;

this.name=name;

}

@Override

publicvoidrun(){

Jedisjedis=null;

try{

jedis=jedisPool.getResource();

//订阅频道为name

jedis.subscribe(subscriber,name);

}catch(Exceptione){

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{

JedisPooljedisPool=newJedisPool("192.168.163.155");

//向ldc频道发布消息

Publisherpublisher=newPublisher(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{

JedisPooljedisPool=newJedisPool("192.168.163.155",6379);

Subscribersubscriber=newSubscriber("黎杜");

//订阅ldc频道

SubThreadthread=newSubThread(jedisPool,subscriber,"ldc");

thread.start();

Thread.sleep(600000);

//取消订阅

subscriber.unsubscribe("ldc");

}

}

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

关键词:Redis 发布 订阅

最近更新