簡(jiǎn)介

Redis發(fā)布與發(fā)布功能（Pub/Sub）是基于事件座位基本的通信機(jī)制，是目前應(yīng)用比較普遍的通信模型，它的目的主要是解除消息的發(fā)布者與訂閱者之間的耦合關(guān)系。

Redis作為消息發(fā)布和訂閱之間的服務(wù)器，起到橋梁的作用，在Redis里面有一個(gè)channel的概念，也就是頻道，發(fā)布者通過(guò)指定發(fā)布到某個(gè)頻道，然后只要有訂閱者訂閱了該頻道，該消息就會(huì)發(fā)送給訂閱者，原理圖如下所示：

Redis同時(shí)也可以使用list類(lèi)型實(shí)現(xiàn)消息隊(duì)列（消息隊(duì)列的實(shí)現(xiàn)以及應(yīng)用場(chǎng)景會(huì)在下一篇文章繼續(xù)講解）。

Redis的發(fā)布與訂閱的功能應(yīng)用還是比較廣泛的，它的應(yīng)用場(chǎng)景有很多。比如：最常見(jiàn)的就是實(shí)現(xiàn)實(shí)時(shí)聊天的功能，還是有就是博客的粉絲文章的推送，當(dāng)博主推送原創(chuàng)文章的時(shí)候，就會(huì)將文章實(shí)時(shí)推送給博主的粉絲。

簡(jiǎn)介完Redis的發(fā)布于訂閱功能，下面就要來(lái)實(shí)操一下，包括linux命令的實(shí)操和java代碼的實(shí)現(xiàn)。

命令實(shí)操

這里就假設(shè)各位讀者都已經(jīng)安裝好自己的虛擬機(jī)環(huán)境和Redis了，若是沒(méi)有安裝好的，可以參考這一篇博文：https://www.cnblogs.com/ zuidongfeng/p/8032505.html

我這里是已經(jīng)安裝好了Redis了，直接啟動(dòng)我們的Redis，我已經(jīng)設(shè)置好了開(kāi)機(jī)啟動(dòng)，上面的那篇博文有講解怎么設(shè)置開(kāi)機(jī)啟動(dòng)。

發(fā)布消息

Redis中發(fā)布消息的命令是publish，具體使用如下所示：

PUBLISH test "haha"：test表示頻道的名稱(chēng)，haha表示發(fā)布的內(nèi)容，這樣就完成了一個(gè)一個(gè)消息的發(fā)布，后面的返回（integer）0表示0人訂閱。

訂閱頻道

于此同時(shí)再啟動(dòng)一個(gè)窗口，這個(gè)窗口作為訂閱者，訂閱者的命令subscribe，使用SUBSCRIBE test就表示訂閱了test這個(gè)頻道訂閱后返回的結(jié)果中由三條信息，第一個(gè)表示類(lèi)型、第二個(gè)表示訂閱的頻道，第三個(gè)表示訂閱的數(shù)量。接著在第一個(gè)窗口進(jìn)行發(fā)布消息：

可以看到發(fā)布者發(fā)布的消息，訂閱者都會(huì)實(shí)時(shí)的接收到，并發(fā)訂閱者收到的信息中也會(huì)出現(xiàn)三條信息，分別表示：返回值的類(lèi)型、頻道名稱(chēng)、消息內(nèi)容。

取消訂閱

若是想取消之前的訂閱可以使用unsubscribe命令，格式為：

unsubscribe??頻道名稱(chēng)
//?取消之前訂閱的test頻道
unsubscribe??test

輸入命令后，返回以下結(jié)果：

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

它分別表示：返回值的類(lèi)型、頻道的名稱(chēng)、該頻道訂閱的數(shù)量。

按模式訂閱

除了直接以特定的名城進(jìn)行訂閱，還可以按照模式進(jìn)行訂閱，模式的方式進(jìn)行訂閱可以一次訂閱多個(gè)頻道，按照模式進(jìn)行訂閱的命令為psubscribe，具體格式如下：

psubscribe??模式
//?表示訂閱名稱(chēng)以ldc開(kāi)頭的頻道
psubscribe??ldc*

輸入上面的命令后，返回如下結(jié)果：

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

這個(gè)也是非常簡(jiǎn)單，分別表示：返回的類(lèi)型（表示按模式訂閱類(lèi)型）、訂閱的模式、訂閱數(shù)。

取消按模式訂閱

假如你想取消之前的按模式訂閱，可以使用punsubscribe來(lái)取消，具體格式：

punsubscribe?模式
//?取消頻道名稱(chēng)按照l(shuí)dc開(kāi)頭的頻道
punsubscribe?ldc*

他的返回值，如下所示：

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

這個(gè)就不多說(shuō)了，表示的意思和上面的一樣，可以看到上面的命令都是有規(guī)律的訂閱SUBSCRIBE，取消就是UNSUBSCRIBE，前面加前綴UN，按模式訂閱也是。

查看訂閱消息

（1）你想查看某一個(gè)模式下訂閱數(shù)是大于零的頻道，可以使用如下格式的命令進(jìn)行操作：

pubsub?channels?模式
//?查看頻道名稱(chēng)以ldc模式開(kāi)頭的訂閱數(shù)大于零的頻道
pubsub?channels?ldc*

（2）假如你想查看某一個(gè)頻道的訂閱數(shù)，可以使用如下命令：

pubsub?numsub?頻道名稱(chēng)

（3）查看按照模式的訂閱數(shù)，可以使用如下命令進(jìn)行操作：

pubsub?numpat

到這里以上的命令操作就基本結(jié)束了，下面就來(lái)代碼實(shí)戰(zhàn)。

代碼實(shí)練

（1）首先第一步想要操作Redis，再SpringBoot項(xiàng)目中引入jedis的依賴(lài)，畢竟jedis是官方推薦使用操作Redis的工具。

????redis.clients
????jedis
????2.9.0

（2）然后創(chuàng)建發(fā)布者Publisher，用于消息的發(fā)布，具體代碼如下：

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;

/**
?*?發(fā)布者
?*?@author?liduchang
?*
?*/
public?class?Publisher?extends?Thread{
?//?連接池?
?private?final?JedisPool?jedisPool;
?//?發(fā)布頻道名稱(chēng)
?private?String?name;
?
?public?Publisher(JedisPool?jedisPool,?String?name)?{
??super();
??this.jedisPool?=?jedisPool;
??this.name?=?name;
?}
?
?@Override
?public?void?run()?{
??//?獲取要發(fā)布的消息
??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))?{
?????//?發(fā)布消息
?????resource.publish(name,?"發(fā)布者:"+Thread.currentThread().getName()+"發(fā)布消息："+message);
????}?else?{
?????break;
????}
???}?catch?(IOException?e)?{
????e.printStackTrace();
???}
??}
?}
}

（3）接著創(chuàng)建訂閱類(lèi)Subscriber，并且繼承JedisPubSub 類(lèi)，重寫(xiě)onMessage、onSubscribe、onUnsubscribe三個(gè)方法，這三個(gè)方法的調(diào)用時(shí)機(jī)在注釋上都有說(shuō)明，具體的實(shí)現(xiàn)代碼如下：

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?{
?//訂閱頻道名稱(chēng)
?private?String?name;
?
?public?Subscriber(String?name)?{
??this.name?=?name;
?}

?/**
??*?訂閱者收到消息時(shí)會(huì)調(diào)用
??*/
?@Override
?public?void?onMessage(String?channel,?String?message)?{
??//?TODO?Auto-generated?method?stub
??super.onMessage(channel,?message);
??System.out.println("頻道："+channel+"??接受的消息為："+message);
?}

?/**
??*?訂閱了頻道會(huì)被調(diào)用
??*/
?@Override
?public?void?onSubscribe(String?channel,?int?subscribedChannels)?{
??System.out.println("訂閱了頻道:"+channel+"??訂閱數(shù)為："+subscribedChannels);
?}

?/**
??*?取消訂閱頻道會(huì)被調(diào)用
??*/
?@Override
?public?void?onUnsubscribe(String?channel,?int?subscribedChannels)?{
??System.out.println("取消訂閱的頻道："+channel+"??訂閱的頻道數(shù)量為："+subscribedChannels);
?}
}

（4）這次創(chuàng)建的才是真正的訂閱者SubThread，上面的Subscriber是指為了測(cè)試實(shí)訂閱的時(shí)候或者發(fā)布消息，能夠有信息輸出：

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）后面就是測(cè)試了，分別測(cè)試發(fā)布與訂閱的測(cè)試，發(fā)布者為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頻道發(fā)布消息
??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");
?}
}

這里為了測(cè)試方便就直接創(chuàng)建線程的方式，更好的話可以使用線程池的方式通過(guò)線程池的submit方法來(lái)執(zhí)行線程，若是不用了可以使用shutdown方式關(guān)閉。

好了這一期的Redis的實(shí)現(xiàn)訂閱與發(fā)布的講解就說(shuō)完了，我們下一期在講解Redis的集群的知識(shí)，下期再見(jiàn)。

特別推薦一個(gè)分享架構(gòu)+算法的優(yōu)質(zhì)內(nèi)容，還沒(méi)關(guān)注的小伙伴，可以長(zhǎng)按關(guān)注一下：

長(zhǎng)按訂閱更多精彩▼
如有收獲，點(diǎn)個(gè)在看，誠(chéng)摯感謝