发布与订阅简介
命令
- SUBSCRIBE: 订阅一个频道
SUBSCRIBE channel [channel ...]
- SUBSCRIBE: 向一个频道发送信息
PUBLISH channel message
- UNSUBSCRIBE: 取消订阅一个频道
UNSUBSCRIBE [channel [channel ...]]
- PSUBSCRIBE:订阅一个或多给定模式的频道
PSUBSCRIBE pattern [pattern ...]
- PUNSUBSCRIBE: 取消订阅一个或多个给定模式的频道
PUNSUBSCRIBE [pattern [pattern ...]]
示例
127.0.0.1:6379> SUBSCRIBE chan1
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "chan1"
3) (integer) 1
127.0.0.1:6379> PUBLISH chan1 "Hello World!"
(integer) 1
1) "message"
2) "chan1"
3) "Hello World!"
127.0.0.1:6379> UNSUBSCRIBE chan1
Reading messages... (press Ctrl-C to quit)
1) "unsubscribe"
2) "chan1"
3) (integer) 0
127.0.0.1:6379> PSUBSCRIBE chan*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "chan*"
3) (integer) 1
1) "pmessage"
2) "chan*"
3) "chan1"
4) "Hello World!"
实现原理
频道的订阅与退订
- 当一个客户端执行了SUBSCRIBE命令订阅某个或者某些频道时，这个客户端与被订阅频道之间建立起一个订阅关系.
redis 将所有频道的订阅关系保存在pubsub_channels字典中，这个字典的键是某个被订阅的频道,值是一个链表，记录了所有订阅这个频道的客户端。
struct redisServer {// ...//保存所有频道的订阅关系dict *pubsub_channels;// ...
};
订阅频道
- 每当客户端执行SUBSCRIBE命令订阅某个或某些频道的时候，服务器都会将客户端与被订阅的频道在pubsub_channels字典中进行关联
- 伪代码
def subscribe(*all_input_channels):#遍历输入的所有频道for channel in all_input_channels:#如果channel不存在于pubsub_channels字典（没有任何订阅者）#那么在字典中添加channel键，并设置它的值为空链表if channel not in server.pubsub_channels:server.pubsub_channels[channel] = []#将订阅者添加到频道所对应的链表的末尾server.pubsub_channels[channel].append(client)
退订频道
- UNSUBSCRIBE命令的行为和SUBSCRIBE命令的行为正好相反，当一个客户端退订某个或某些频道的时候，服务器将从pubsub_channels中解除客户端与被退订频道之间的关联
- 伪代码
def unsubscribe(*all_input_channels):#遍历要退订的所有频道for channel in all_input_channels:#在订阅者链表中删除退订的客户端server.pubsub_channels[channel].remove(client)#如果频道已经没有任何订阅者了（订阅者链表为空）#那么将频道从字典中删除if len(server.pubsub_channels[channel]) == 0:server.pubsub_channels.remove(channel)
模式的订阅与退订
- 与频道订阅类似,服务器也将所有模式的订阅关系都保存在服务器状态的pubsub_patterns属性里面
模式订阅
每当客户端执行PSUBSCRIBE命令订阅某个或某些模式的时候，服务器会对每个被订阅的模式执行以下两个操作：
- 将客户端与被订阅的模式在pubsub_patterns字典中进行关联
- 遍历所有已经存在的键，检查这些键是否与被订阅的模式匹配。如果某个键与模式匹配，那么服务器就会将这个客户端添加到与这个键相关的订阅者链表的末尾
- 伪代码
def psubscribe(*all_input_patterns):#遍历输入的所有模式for pattern in all_input_patterns:#创建新的pubsubPattern结构#记录被订阅的模式，以及订阅模式的客户端pubsubPattern = create_new_pubsubPattern()pubsubPattern.client = clientpubsubPattern.pattern = pattern#将新的pubsubPattern追加到pubsub_patterns链表末尾server.pubsub_patterns.append(pubsubPattern)
模式退订
- 伪代码
def punsubscribe(*all_input_patterns):# 遍历所有要退订的模式for pattern in all_input_patterns:# 遍历pubsub_patterns链表中的所有pubsubPattern结构for pubsubPattern in server.pubsub_patterns:# 如果当前客户端和pubsubPattern记录的客户端相同# 并且要退订的模式也和pubsubPattern记录的模式相同if client == pubsubPattern.client and \pattern == pubsubPattern.pattern:# 那么将这个pubsubPattern从链表中删除server.pubsub_patterns.remove(pubsubPattern)
发送消息
当一个Redis客户端执行PUBLISH＜channel＞＜message＞命令将消息message发送给频道channel的时候，服务器需要执行以下两个动作:
- 1）将消息message发送给channel频道的所有订阅者。
- 2）如果有一个或多个模式pattern与频道channel相匹配，那么将消息message发送给pattern模式的订阅者
- 伪代码
def channel_publish(channel, message):#如果channel键不存在于pubsub_channels字典中#那么说明channel频道没有任何订阅者#程序不做发送动作，直接返回if channel not in server.pubsub_channels:return#运行到这里，说明channel频道至少有一个订阅者#程序遍历channel频道的订阅者链表#将消息发送给所有订阅者for subscriber in server.pubsub_channels[channel]:send_message(subscriber, message)
查看订阅信息
- PUBSUB CHANNELS[pattern]子命令用于返回服务器当前被订阅的频道
- PUBSUB NUMSUB[channel-1 channel-2…channel-n]子命令接受任意多个频道作为输入参数，并返回这些频道的订阅者数量。
- PUBSUB NUMPAT子命令用于返回服务器当前被订阅模式的数量。
重点回顾
- 服务器状态在pubsub_channels字典保存了所有频道的订阅关系：SUBSCRIBE命令负责将客户端和被订阅的频道关联到这个字典里面，而UNSUBSCRIBE命令则负责解除客户端和被退订频道之间的关联。
- 服务器状态在pubsub_patterns链表保存了所有模式的订阅关系：PSUBSCRIBE命令负责将客户端和被订阅的模式记录到这个链表中，而PUNSUBSCRIBE命令则负责移除客户端和被退订模式在链表中的记录。
- PUBLISH命令通过访问pubsub_channels字典来向频道的所有订阅者发送消息，通过访问pubsub_patterns链表来向所有匹配频道的模式的订阅者发送消息。
- PUBSUB命令的三个子命令都是通过读取pubsub_channels字典和pubsub_patterns链表中的信息来实现的。