Redis的发布/订阅工作模式详解

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的控制台输出如下图所示：