redis发布订阅模型

文章目录

发布与订阅

Redis的发布订阅基于publish,subscribe,psubscribe

订阅

subscribe "news.it"

发布

publish "news.it" "hello"

订阅

image-20230728101944396

发布

image-20230728102011059

订阅的状态的推进

image-20230728102744213

向new.it发送消息

image-20230728102755236

1.频道订阅和退订

Redis订阅关系保存在服务器状态的pubsub_channels字典

key为channel, value为链表，链表上有订阅该频道的Client

struct redisServer{
  //...
  
  //保存频道的订阅关系
  
  dict *pubsub_channels
  //...
}

如图所示

image-20230728141631387

1.1订阅频道subscribe

Client订阅pubsub_channels有两种情况

  • 频道有其他订阅者,把value添加到链表尾部
  • 没有订阅者，就把先创建key再添加value
subscribe "news.sport" "news.movie"

image-20230728144804639

伪代码
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) = (1
		＃ 将订阅者添加到频道所对应的链表的未尾
		server.pubsub channels [channel].append (client)
1.2 退订频道
unsubscribe "news.sport"  "news.movie"

image-20230728145128522

伪代码
def unsubscribe (*all input channels):
		＃遍历要退订的所有频道
		for channel in all input channels :
			＃ 在订阅者链表中副除退订的客户端
			server . pubsub channels [channel1. remove (client)
			# 如果频道已经没有任何订阅者了（订间者链表为空）
			＃ 那么将频道从字典中刷除
			if len (server.pubsub channels [channel]) == 0 :
				server.pubsub channels. remove (channel)

2.模式订阅和退订

struct redisServer{
  // ...
  //保持所有模式订阅关系
  list *pubsub_patterns;
  // ...
}

pubsub_patterns是链表,每个节点都包括一个pubsub_Patterns结构

typedef struct pubsubPattern{
  //订阅模式Client
  redisClient *client;
  //被订阅的模式
  robj *pattern;
}

示意图

image-20230728153722027

client-7,8,9分别在订阅 music.*,book.*,news.*

image-20230728154159308

2.1订阅模式psubscribe

客户端执行psubscribe，服务端会执行两个操作

  1. 新建pubsubPattern，讲结构的pattern设置为被订阅的模式
  2. 将pubsubPattern添加到pubsub_patterns链表尾部

执行psubscribe之前的pubsub_patterns

image-20230728155257011

执行psubscribe之后的pubsub_patterns

def subscribe (*all_ input patterns):
 ＃遍历输入的所有模式
	for pattern in all input patterns:
		#创建新的 pubsubPattern 结构
   # 记录被订间的模式，以及订阅模式的容户端
   pubsubPattern = create new pubsubPattern (
   pubsubPattern.client= client
   pubsubPattern.pattern= pattern
   # 将新的 pubsubPattern 追加到pubsubpatterns链表末尾
   server.pubsub patterns.append (pubsubPattern)
2.2 退订模式

punsubscribe

image-20230728160920126

当执行退订模式的时候，相应的pubsubPattern结构会被删除

执行punsubscribe "news.*"之前的pubsub_patterns

image-20230728161636374

执行punsubscribe "news.*"之后的pubsub_patterns

image-20230728161700300

def unsubscribe (*all input patterns) :
		＃ 遍历所有要退订的模式
		for pattern in all input patterns:
				# 遍历 pubsub patterns 链表中的所有pubsubPattern 结构
				for pubsubPattern in server.pubsub patterns:
							#如果当前客户端和pubsuibPattern 记录的客户端相同
							#并且要退订的模式也和 pubsubPattern 记录的模式相同
							if client == pubsubPattern.client and \
								pattern == pubsubPattern.pattern:
                # 那么将这个pubsubPattern 从链表中利除
                server.pubsub patterns.remove (pubsubPattern)

3.发送消息

PUBLISH <channel> <message>

将message发送到channel的订阅者

一个或多个pattern与频道channel匹配

3.1 消息发送频道订阅者

image-20230728162209425

PUBLISH "news.it" "hello"		// client-1,client-2,client-3都会收到
伪代码
def channel_publish (channel, message) :
# 如果 channel键不存在于pubsub channels 字典中
# 那么说明 channe1 频道没有任何订阋者
# 程序不做发送动作，直接返回
if channel not in server.pubsub channels:
	return
    
# 运行到这里，说明 channel 频道至少有一个订阅者
# 程序遍历 channe1 频道的订阅者链表
＃ 将消,息发送给所有订阅者
for subscriber in server.pubsub channels [channel] :
	send message (subscriber, message)
3.2消息发送模式订阅者

image-20230728162832130

PUBLISH "news.it" "hello"		// 和news.*匹配，news.*会收到
伪代码
def pattern_publish (channel, message):
	# 遍历所有模式订阅消息
	for pubsubPattern in server.pubsub patterns:
		＃ 如果频道和模式相匹配
		if match (channel, pubsubPattern.pattern)：
			#那么格消息发洪给订回该模式的容户瑞
			send message (pubsubPattern.client, message)

总结的Publish

def publish (channel, message):
		＃ 将消息发送给 channe1 频道的所有订间者
		channel publish (channel, message)
				#将消息发送给所有和 channel 頻道相匹配的模式的订回者
				pattern publish (channel, message)

4.查看订阅消息

PUBSUB查看频道或者模式相关消息

4.1 pubsub channels
pubsub channels <pattern>(可选)  //pattern可选，选择表示返回匹配的频道

def pubsub channels (pattern=None) :
	#一个列表，用于记录所有符合条件的频道
	channel_list = []
    
	# 遍历服务器中的所有频道
	#（也即是pubsub channels 字典的所有键）
	for channel in server.pubsub channels:
		# 当以下两个条件的任意一个满足时，将频道添加到链表里面：
		#1 ）用户没有指定pattezn 参数
		#2 ）用户指定了pattern 参数，并且channel 和pattern 匹配
		if (pattern is None) or match (channel, pattern):
		channel list.append (channel)
# 向客户端返回频道列表
return channel_list

image-20230728164616065

被订阅的4个频道

redis> PUBSUB CHANNELS
1)"news. it"
2)"news.sport"
3)"news. business"
4)"news.movie"

进行频道pattern筛选

redis> PUBSUB CHANNELS "news. [is]*"	
  //返回所有名称以"news."开头，并且后面紧跟着一个或多个字母"i"或"s"的频道的列表
1)"news.it"
2)"news. sport"
4.2 pubsub numsub
pubsub numsub [channel-1 channel-2 ...channel-n]  //接受多个频道作为入参，返回这些频道的订阅者
伪代码
def pubsub numsub (*all input channels) :
	＃ 遍历输入的所有频道
	for channel in all input channels:
		#如果 pubsub channels 宇典中没有 channe1 这个键
		＃那么说明 channel 频道没有任何订回者
		if channel not in server.pubsub channels:
			#返回频道名
      reply channel name (channel)
      # 订阅者数量为。
      reply subscribe count (0)
#如果 pubsub channels 字典中存在 channe1 键
# 那么说明channel 频道至少有一个订阅者
else:
      #返回频道名
      reply channel name (channel)
      ＃ 订阅者链表的长度就是订用者数量
      reply subscribe count (len (server .pubsub_channels [channel]))

image-20230728170459519

redis> PUBSUB NUMSUB news.it news.sport news.business news.movie
1)"news. it"
2)"3"
3)"news. sport"
4)"2"
5)"news.business"
6)"2"
7)"news. movie"
8)"1"
4.3 pubsub numpat

返回服务器当前被订阅模式的数量

伪代码
def pubsub_numpat():
		# pubsub_patterns 链表长度是被订阅数量
reply pattern count (len(server.pubsub patterns))

image-20230728170821668

对于上方链表来说，执行pubsub numpat结果

redis> pubsub numpat
(integer) 3

5.回顾

  1. 服务器在pubsub_channels保存频道订阅关系，subscribe和unsubscribe
  2. 服务器在pubsub_patterns保存模式订阅关系,psubscribe和punsubscribe
  3. publish发送消息
  4. pubsub读取pubsub_channels和pubsub_patterns来实现