簡(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類型實(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表示頻道的名稱，haha表示發(fā)布的內(nèi)容，這樣就完成了一個(gè)一個(gè)消息的發(fā)布，后面的返回（integer）0表示0人訂閱。

訂閱頻道

于此同時(shí)再啟動(dòng)一個(gè)窗口，這個(gè)窗口作為訂閱者，訂閱者的命令subscribe，使用SUBSCRIBE test就表示訂閱了test這個(gè)頻道

訂閱后返回的結(jié)果中由三條信息，第一個(gè)表示類型、第二個(gè)表示訂閱的頻道，第三個(gè)表示訂閱的數(shù)量。接著在第一個(gè)窗口進(jìn)行發(fā)布消息：

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

取消訂閱

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

unsubscribe  頻道名稱  
// 取消之前訂閱的test頻道  
unsubscribe  test 

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

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

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

按模式訂閱

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

psubscribe  模式  
// 表示訂閱名稱以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)單，分別表示：返回的類型（表示按模式訂閱類型）、訂閱的模式、訂閱數(shù)。

取消按模式訂閱

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

punsubscribe 模式  
// 取消頻道名稱按照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 模式  
// 查看頻道名稱以ldc模式開(kāi)頭的訂閱數(shù)大于零的頻道  
pubsub channels ldc* 

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

pubsub numsub 頻道名稱 

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

pubsub numpat 

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

代碼實(shí)練

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

<dependency>  
    <groupId>redis.clients</groupId>  
    <artifactId>jedis</artifactId>  
    <version>2.9.0</version>  
</dependency> 

（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ā)布頻道名稱  
 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)建訂閱類Subscriber，并且繼承JedisPubSub 類，重寫(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 {  
 //訂閱頻道名稱  
 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ā)布者為T(mén)estPublisher，訂閱者為T(mén)estSubscriber： 

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)閉。