Redis的SUBSCRIBE、UNSUBSCRIBE和PUBLISH命令實現了消息的發布/訂閱功能.發送者(消息發布者)不必要編程,就能夠向特定的接收者(消息訂閱者)發送消息了.Redis會將已發布的消息放入指定的頻道之中,消息發布者不必要知道具體有哪些消息訂閱者.訂閱者可能會訂閱一個或多個頻道,并且只能接收已訂閱頻道中的消息,它們也不必要知道具體有哪些消息發布者.這種方式能夠充分解耦發布者和訂閱者之間的關系,也使得網絡拓撲具有更好的伸縮性和動態性.

例如,若客戶端想要訂閱頻道foo和bar,它可以運行SUBSCRIBE命令,以頻道名稱作為參數,如下所示：

SUBSCRIBE foo bar

如果其他客戶端向這些頻道發送消息,那么Redis就會將這些消息推送給所有的訂閱客戶端.

如果一個客戶端訂閱了一個或多個頻道,那么除了訂閱和退訂命令之外,這個客戶端就不能運行其他命令了.訂閱和退訂操作的應答信息是以消息的形式發送的,客戶端只能讀取與其相關的消息流,每個消息的第一個部分表現消息的類型.處于訂閱狀態的客戶端只能運行SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE、PUNSUBSCRIBE、PING和QUIT命令.

一、環境描述

1. 主機配置

   CPU：單核

   內存：2 GB

   IP：192.168.1.109

2. 操作系統

   CentOS 6.6 x86_64 Minimal

3. Redis版本

   Redis server v=3.2.4 sha=00000000:0 malloc=jemalloc-4.0.3 bits=64 build=431d8e4a684c794b

4. Redis安裝方式

   依照《在CentOS上安裝Redis緩存系統》

二、發布/訂閱的相關命令

1. SUBSCRIBE

使得客戶端訂閱指定的頻道.一旦客戶端進入訂閱狀態,它就不克不及運行除了SUBSCRIBE、PSUBSCRIBE、UNSUBSCRIBE和PUNSUBSCRIBE命令之外的其他命令.

這個命令的運行格式如下所示：

SUBSCRIBE channel [channel ...]

時間復雜度為O(N),N是客戶端想要訂閱的頻道的數量.

2. UNSUBSCRIBE

使得客戶端退訂指定的頻道,如果不指定任何頻道,那么就退訂所有頻道.

這個命令的運行格式如下所示：

UNSUBSCRIBE [channel [channel ...]]

時間復雜度為O(N),N是客戶端想要退訂的頻道的數量.

3. PSUBSCRIBE

使得客戶端訂閱指定模式的頻道,支持glob風格的模式,例如：

• h?llo：可以訂閱hello、hallo和hxllo頻道(?表現單個任意字符).

• h*llo：可以訂閱hllo和heeeello頻道(*表示任意多個任意字符,包含空字符).

• h[ae]llo：可以訂閱hello和hallo頻道,但是不克不及訂閱hillo頻道(選擇[和]之間的任意一個字符).

如果想要將上述通配符作為普通字符進行處理,則必要使用\符號進行轉義.

這個命令的運行格式如下所示：

PSUBSCRIBE pattern [pattern ...]

時間復雜度為O(N),N是客戶端已經訂閱的模式的數量.

4. PUNSUBSCRIBE

使得客戶端退訂指定的模式所對應的頻道,如果不指定任何模式,那么就退訂所有的模式.

這個命令的運行格式如下所示：

PUNSUBSCRIBE [pattern [pattern ...]]

時間復雜度為O(N+M),N是客戶端已經訂閱的模式的數量,M是系統中已經訂閱的模式的總數量..

5. PUBLISH

向指定的頻道提交一條消息.返回值是一個整數,表現接收到這條消息的客戶端的數量.

這個命令的運行格式如下所示：

PUBLISH channel message

時間復雜度為O(N+M),N是訂閱這個接收頻道的客戶端的數量,M是系統中已經訂閱的模式的總數量.

6. PUBSUB

PUBSUB命令是一個自檢命令,可用于檢查發布/訂閱子系統的狀態.這個命令是由幾個子命令組成的,下面會分別描述.這個命令的一般格式如下所示：

PUBSUB <subcommand> ... args ...

6.1 PUBSUB CHANNELS [pattern]

列出當前有效的頻道.有效頻道是具有一個或多個訂閱者(不包含訂閱模式的客戶端)的發布/訂閱頻道.

如果沒有指定任何模式,那么就會列出所有的頻道.如果指定了模式,那么就只會列出匹配這個模式的所有頻道(此處使用glob風格的模式).

這個命令的返回值是一個數組,它會列出所有的有效頻道,包含匹配指定模式的有效頻道.

時間復雜度為O(N),N是有效頻道的數量,假設模式匹配的時間是恒定的(相對于較短的頻道名稱和模式而言).

6.2 PUBSUB NUMSUB [channel-1 … channel-N]

返回指定頻道的訂閱者的數量(不會計算訂閱模式的客戶端的數量).

這個命令的返回值是一個數組,它會列出參數指定的所有頻道,以及每個頻道的訂閱者的數量.返回值的格式為頻道、數量、頻道、數量、...,因此,這個列表是扁平的.返回值列出的頻道順序和命令調用時指定的頻道順序是相同的.注意,調用這個命令時可以不指定頻道,此時返回值是一個空列表.

時間復雜度為O(N),N是命令中指定的頻道的數量.

6.3 PUBSUB NUMPAT

返回模式的訂閱數量(也便是所有客戶端運行PSUBSCRIBE命令的總次數).注意,這個數量不是訂閱模式的客戶端的數量,而是所有客戶端訂閱的模式的總數量.

這個命令的返回值是一個整數,表現所有客戶端訂閱的模式的總數量.

時間復雜度為O(1).

三、推送消息的格式

Redis的發布/訂閱功能有兩種工作模式：訂閱頻道(channel)和訂閱模式(pattern).這兩種工作模式推送消息的格式是分歧的,如下文所述.

1. 訂閱頻道

• subscribe消息

  這種消息表現客戶端已經成功地訂閱了指定的頻道,它由三部分組成：

  第一部分是subscribe字符串；第二部分表現想要訂閱的頻道名稱；第三部分表現客戶端當前已經訂閱的頻道數量.消息格式如下圖所示：

  

• unsubscribe消息

  這種消息表現客戶端已經成功地退訂了指定的頻道,它由三部分組成：

  第一部分是unsubscribe字符串；第二部分表現想要退訂的頻道名稱；第三部分表現客戶端當前已經訂閱的頻道數量.若第三部分的值為零,則表現客戶端沒有訂閱任何頻道,此時客戶端便可以運行任意種類的Redis命令了.消息格式如下圖所示：

  

• message消息

  這種消息表現訂閱頻道的客戶端已經成功地收到了另一個客戶端向這個頻道發送的信息,它由三部分組成：

  第一部分是message字符串；第二部分表現推送消息的頻道名稱；第三部分表現另一個客戶端向這個頻道發送的消息內容.消息格式如下圖所示：

  

2. 訂閱模式

• psubscribe消息

  這種消息表現客戶端已經成功地訂閱了指定的模式,它由三部分組成：

  第一部分是psubscribe字符串；第二部分表現想要訂閱的模式名稱；第三部分表現客戶端當前已經訂閱的模式數量.消息格式如下圖所示：

  

• punsubscribe消息

  這種消息表現客戶端已經成功地退訂了指定的模式,它由三部分組成：

  第一部分是punsubscribe字符串；第二部分表現想要退訂的模式名稱；第三部分表現客戶端當前已經訂閱的模式數量.若第三部分的值為零,則表現客戶端沒有訂閱任何模式,此時客戶端便可以運行任意種類的Redis命令了.消息格式如下圖所示：

  

• pmessage消息

  這種消息表現訂閱模式的客戶端已經成功地收到了另一個客戶端向這個模式所對應的頻道發送的信息,它由三部分組成：

  第一部分是pmessage字符串；第二部分表現推送消息的頻道模式的名稱；第三部分表現另一個客戶端向這個頻道模式發送的消息內容.消息格式如下圖所示：

  

四、注意事項

1. 數據庫和作用域

發布/訂閱和鍵空間沒有任何關系.它被設計為不會對鍵空間造成任何影響,包含數據庫編號.這就意味著,在db 10上發布消息,仍然可以被在db 1上的訂閱者監聽到.

如果你必要某種類型的作用域,那么可以為頻道名稱添加環境前綴,例如：test、staging、production,等等.

2. 頻道退訂

如果客戶端是在Redis命令行(redis-cli)中進入訂閱監聽狀態的,那么它是不能直接運行UNSUBSCRIBE或PUNSUBSCRIBE命令的,必需通過telnet之類的工具才能在訂閱監聽狀態中進行退訂操作.

3. 訂閱計數

在subscribe、unsubscribe、psubscribe和punsubscribe消息類型中,消息的最后一部門是客戶端仍然有效的訂閱數量.這個數量實際上是客戶端仍然在訂閱的頻道和模式的總數.只有當這個數量變為零時,客戶端才會退出發布/訂閱狀態,這就意味著客戶端已經退訂了所有的頻道和模式

4. 模式訂閱和頻道訂閱

如果某個客戶端訂閱了多個模式(或者多個模式和頻道),而且這些模式都能匹配到同一條消息,那么這個客戶端就會多次收到這條相同的消息.例如,某個客戶端同時訂閱了一個頻道和一個模式：

SUBSCRIBE foo

PSUBSCRIBE f*

在上面的例子中,如果向頻道foo發送一條消息,那么這個客戶端將會收到兩條消息：一條是message類型的消息,另一條是pmessage類型的消息.

五、命令行示例

1. 訂閱頻道

Step-1 訂閱頻道

打開一個Shell終端(此處取名為終端-1),運行以下命令：

telnet localhost 6379

然后,在telnet提示符中輸入以下命令：

subscribe mychannel

若上述命令的返回信息(也便是subscribe消息)如下圖所示,則表示頻道訂閱成功：

在上圖中,*3表示消息有三部分組成；$9表示下面有9字節長的字符串,也便是subscribe字符串,這是消息的第一部分；接下來,還有一個$9,表示下面有9字節長的字符串,也便是mychannel字符串,這是消息的第二部分；最后,:1表示這個客戶端訂閱的頻道數量.在下面的示例中,返回消息的結構和含義大致相同,本文也就不再贅述.

Step-2 發布消息

打開另一個Shell終端(此處取名為終端-2),運行以下命令：

redis-cli

進入Redis客戶端的命令行之后,運行以下命令,發布一條消息：

publish mychannel hello

若上述命令在終端-2中的返回信息如下圖所示,則表現消息發送成功：

在上圖中,(integer) 1表現收到這條消息的客戶端的數量.

此時,在終端-1中可以看到客戶端收到的消息(也便是message消息),如下圖所示：

Step-3 退訂頻道

在終端-1的telnet提示符中輸入以下命令：

unsubscribe mychannel

若上述命令的返回信息(也便是unsubscribe消息)如下圖所示,則表示頻道退訂成功：

2. 訂閱模式

Step-1 訂閱模式

打開一個Shell終端(此處取名為終端-1),運行以下命令：

telnet localhost 6379

然后,在telnet提示符中輸入以下命令：

psubscribe mychannel.*

若上述命令的返回信息(也便是psubscribe消息)如下圖所示,則表示模式訂閱成功：

Step-2 發布消息

打開另一個Shell終端(此處取名為終端-2),運行以下命令：

redis-cli

進入Redis客戶端的命令行之后,運行以下命令,發布一條消息：

publish mychannel.test hello123

若上述命令在終端-2中的返回信息如下圖所示,則表現消息發送成功：

在上圖中,(integer) 1表現收到這條消息的客戶端的數量.

此時,在終端-1中可以看到客戶端收到的消息(也便是pmessage消息),如下圖所示：

Step-3 退訂模式

在終端-1的telnet提示符中輸入以下命令：

punsubscribe mychannel.*

若上述命令的返回信息(也便是punsubscribe消息)如下圖所示,則表示模式退訂成功：

3. 同時訂閱頻道和模式

Step-1 訂閱頻道和模式

打開一個Shell終端(此處取名為終端-1),運行以下命令：

telnet localhost 6379

然后,在telnet提示符中輸入以下命令：

subscribe foop

subscribe f*

若上述命令的返回信息(也便是psubscribe消息)如下圖所示,則表示頻道和模式訂閱成功：

Step-2 發布消息

打開另一個Shell終端(此處取名為終端-2),運行以下命令：

redis-cli

進入Redis客戶端的命令行之后,運行以下命令,發布一條消息：

publish foo hello

若上述命令在終端-2中的返回信息如下圖所示,則表現消息發送成功：

在上圖中,(integer) 2表現收到這條消息的客戶端的數量.

此時,在終端-1中可以看到客戶端收到兩條消息(也便是message消息和pmessage消息),如下圖所示：

六、Java示例

Jedis是一種用Java語言開發的Redis客戶端,它具有輕量級和功能完備的特點.Jedis完全兼容于Redis 2.8.x和3.0.x版本.

以下代碼簡單示范了如何通過Jedis操作Redis緩存.

• App.java

package org.xninja.ghoulich.JedisTest;

import redis.clients.jedis.Jedis;

public class App {

@SuppressWarnings("resource")

public static void main(String[] args) {

final Jedis jedis = new Jedis("192.168.1.109", 6379);

final Jedis pjedis = new Jedis("192.168.1.109", 6379);

final MyListener listener = new MyListener();

final MyListener plistener = new MyListener();

Thread thread = new Thread(new Runnable() {

public void run() {

jedis.subscribe(listener, "mychannel");

}

});

Thread pthread = new Thread(new Runnable() {

public void run() {

pjedis.psubscribe(plistener, "mychannel.*");

}

});

thread.start();

pthread.start();

}

}

該法式建立了兩個Jedis客戶端,然后又建立了兩個發布/訂閱監聽器,最后啟動了兩個線程,分別用于監聽一個頻道和一個模式.

• MyListener.java

package org.xninja.ghoulich.JedisTest;

import redis.clients.jedis.JedisPubSub;

public class MyListener extends JedisPubSub {

// 取得訂閱的消息后的處理

public void onMessage(String channel, String message) {

System.out.println("onMessage: " + channel + "=" + message);

if (message.equals("quit"))

this.unsubscribe(channel);

}

// 初始化訂閱時候的處理

public void onSubscribe(String channel, int subscribedChannels) {

System.out.println("onSubscribe: " + channel + "=" + subscribedChannels);

}

// 撤消訂閱時候的處理

public void onUnsubscribe(String channel, int subscribedChannels) {

System.out.println("onUnsubscribe: " + channel + "=" + subscribedChannels);

}

// 初始化按表達式的方式訂閱時候的處理

public void onPSubscribe(String pattern, int subscribedChannels) {

System.out.println("onPSubscribe: " + pattern + "=" + subscribedChannels);

}

// 撤消按表達式的方式訂閱時候的處理

public void onPUnsubscribe(String pattern, int subscribedChannels) {

System.out.println("onPUnsubscribe: " + pattern + "=" + subscribedChannels);

}

// 取得按表達式的方式訂閱的消息后的處理

public void onPMessage(String pattern, String channel, String message) {

System.out.println("onPMessage: " + pattern + "=" + channel + "=" + message);

if (message.equals("quit"))

this.punsubscribe(pattern);

}

}

這個監聽器會對頻道和模式的訂閱、接收消息和退訂等變亂進行監聽,然后進行相應的處理.

在Eclipse中運行App.java的main函數,此時會觸發頻道和模式的訂閱變亂,控制臺中的輸出如下圖所示：

此時,示例法式已經訂閱了mychannel頻道和mychannel.*模式.然后,在redis-cli命令行中輸入以下命令,向mychannel頻道和mychannel.*模式各發送一條消息：

publish mychannel "hello message for channel"

publish mychannel.test "hello message for pattern"

此時,會觸發示例程序的接收消息變亂,Eclipse的控制臺輸出如下圖所示：

最后,在redis-cli命令行中輸入以下命令,向mychannel頻道和mychannel.*模式各發送一條用于退訂的消息：

publish mychannel quit

publish mychannel.test quit

由示例法式的源碼可知,當監聽器收到內容為“quit”的消息時,便會退訂mychannel頻道和mychannel.*模式,然后終止執行.Eclipse的控制臺輸出如下圖所示：