参考资料:

《Redis进阶——发布订阅详解》

《Redis 发布订阅》

《Redis进阶 - 消息传递:发布订阅模式详解》

        写在开头:本文为学习后的总结,可能有不到位的地方,错误的地方,欢迎各位指正。

目录

一、什么是发布订阅

二、发布订阅的实现

        1、基于频道的发布订阅

        (1)使用方法

        (2)具体实现

        2、基于模式的发布订阅

        (1)使用方法

        (2) 具体实现

补充

        退订

        发布订阅实际应用


一、什么是发布订阅

        Redis的发布订阅实现了类似mq的消息推送功能。

        假设我们有一个支付接口,在这个接口中,又要去调用其他三个服务。

        

        我们可能会使用上图这样线性调用的方法来执行,但这样的设计会带来几个问题:

  • 下单支付业务与其他业务重度耦合,每当有个新业务需要支付结果,就需要改动下单支付的业务
  • 如果调用业务过多,会导致下单支付接口响应时间变长。另外,如果有任一下游接口响应变慢,就会同步导致下单支付接口响应也变长
  • 如果任一下游接口失败,可能导致数据不一致的情况。比如说下图,先调用 A,成功之后再调用 B,最后再调用 C,当B失败了,就会导致数据不一致

        为此,我们希望能够将这几个功能模块进行拆分、解耦,于是我们便想到了使用中间件消息推送,当支付接口被调用时,利用这个机制通知另外三个服务进行处理。当然,我们可以使用专业的mq如rabbitmq等来实现这个功能,不过Redis也能实现相同的效果,这就是它的发布/订阅机制。        

         发布订阅机制包含2个对象,发布者和订阅者,当两者建立发布订阅关系后,当发布者向redis中发布消息时,订阅者就能收到这个消息,就像上面的业务场景,支付接口只需要向特定频道发送消息,其他下游业务订阅这个频道,就能收相应消息,然后做出业务处理即可

二、发布订阅的实现

        1、基于频道的发布订阅

        (1)使用方法

        在Redis中,我们使用subscribe指令指定当前客户端订阅的频道,一个订阅者可以订阅多个频道,这个频道如果不存在则会先进性创建。

# subscribe channel  [channel ... ]
# 订阅给定的一个或多个频道

127.0.0.1:6379> subscribe meihuashisan meihuashisan2 
Reading messages... (press Ctrl-C to quit)
 
1) "subscribe"    -- 返回值类型:表示订阅成功!
2) "meihuashisan" -- 订阅频道的名称
3) (integer) 1    -- 当前客户端已订阅频道的数量
 
1) "subscribe"
2) "meihuashisan2"
3) (integer) 2

         发布者则使用publish向某个频道发布消息。

# 格式为publish channel message 
# 将消息发送给指定频道 channel
# 返回结果:接收到信息的订阅者数量,无订阅者返回0

127.0.0.1:6379> publish meihuashisan "I am meihuashisan"
(integer) 1  # 接收到信息的订阅者数量,无订阅者返回0

        当发布者发布消息后,订阅了该频道的客户端就会收到该条消息。 

127.0.0.1:6379> subscribe meihuashisan csdn 
Reading messages... (press Ctrl-C to quit)
 
1) "subscribe"    -- 返回值类型:表示订阅成功!
2) "meihuashisan" -- 订阅频道的名称
3) (integer) 1    -- 当前客户端已订阅频道的数量
 
1) "subscribe"
2) "csdn"
3) (integer) 2
 
 
 ---------------------追加变化如下:(实时接收到了该频道的发布者的消息)------------
 
1) "message"           -- 返回值类型:消息
2) "meihuashisan"      -- 来源(从哪个频道发过来的)
3) "I am meihuashisan" -- 消息内容
 

        (2)具体实现

        在redisServer中,有一个字典类型的字段pubsub_channels,用来保存订阅信息,其中key为频道,value为订阅该频道的客户端

struct redisServer { 
  /* General */ 
  pid_t pid; 
 
  //其他省略
    ...
 
  // 将频道映射到已订阅客户端的列表(就是保存客户端和订阅的频道信息)
  dict *pubsub_channels; /* Map channels to list of subscribed clients */ 
}

 ​​​​​​​        ​​​​​​​如下图的这个示例中, client1 、 client2订阅了 channel_1 , 而其他频道也分别被别的客户端所订阅。

        当客户端 client7执行命令subscribe channel1 channel2 channel3 ,那么就会把这个客户端分别加到相应key的链表末尾。

         当调用 publish channel message 命令, 程序首先根据 channel 定位到字典的键, 然后将信息发送给字典值链表中的所有客户端。 以上图为例, 如果某个客户端执行命令 PUBLISH channel1 "hello moto" ,那么 client2 、 client5 和 client1 三个客户端都将接收到 "hello moto" 信息。

 

        2、基于模式的发布订阅

        (1)使用方法

        在基于频道的订阅中,我们通过输入某个频道的完整名称来实现订阅,而基于模式的订阅在使用时不需要指定全名,而是使用一个模式匹配字符串代替,当与该模式匹配的频道有消息发布时,就可以接收到。

        在模式匹配中,通配符中?表示1个占位符,*表示任意个占位符(包括0),?*表示1个以上占位符。例如tweet.shop.*能匹配tweet.shop.ipad和tweet.shop.kindle。

        如下图,client123与client256通过基于模式的发布订阅接收与tweet.shop.*模式匹配的频道的消息。因此当tweet.shop.ipad和tweet.shop.kindle频道有消息发布时,client123与client256都能接收到。

 

        订阅者通过psubscribe pattern  [pattern ...] 进行模式订阅。

#  订阅 “a?” "com.*" 2种模式频道(注意中间的空格表明这是2个模式)
127.0.0.1:6379> psubscribe a? com.*
# 进入订阅状态后处于阻塞,可以按Ctrl+C键退出订阅状态
Reading messages... (press Ctrl-C to quit) 
 
 
1) "psubscribe"  -- 返回值的类型:显示订阅成功
2) "a?"          -- 订阅的模式
3) (integer) 1   -- 目前已订阅的模式的数量
 
 
1) "psubscribe"
2) "com.*"
3) (integer) 2
 
 
# 接收消息 (已订阅 “a?” "com.*" 两种模式!)
 
# ---- 发布者第1条命令: publish ahead "hello"
结果:没有接收到消息,匹配失败,不满足 “a?” ,“?”表示一个占位符, a后面的head有4个占位符
 
 
# 发布者第2条命令:  publish aa "hello" (满足 “a?”)
1) "pmessage" -- 返回值的类型:信息
2) "a?"       -- 信息匹配的模式:a?
3) "aa"       -- 信息本身的目标频道:aa
4) "hello"    -- 信息的内容:"hello"
 
 
# 发布者第3条命令:publish com.juc "hello2"(满足 “com.*”, *表示任意个占位符)
1) "pmessage" -- 返回值的类型:信息
2) "com.*"    -- 匹配模式:com.*
3) "com.juc"  -- 实际频道:com.juc
4) "hello2"   -- 信息:"hello2"
 
# 发布者第4条命令: publish com. "hello3"(满足 “com.*”, *表示任意个占位符)
1) "pmessage" -- 返回值的类型:信息
2) "com.*"    -- 匹配模式:com.*
3) "com."     -- 实际频道:com.
4) "hello3"   -- 信息:"hello3"

         发布者还是通过publish channel message发布消息。

# 1. ahead 不符合“a?”模式,?表示1个占位符
127.0.0.1:6379> publish ahead "hello"  
(integer) 0    -- 匹配失败,0:无订阅者
 
 
# 2. aa 符合“a?”模式,?表示1个占位符
127.0.0.1:6379> publish aa "hello"      
(integer) 1
 
# 3. 符合“com.*”模式,*表示任意个占位符
127.0.0.1:6379> publish com.juc "hello2" 
(integer) 1
 
# 4. 符合“com.*”模式,*表示任意个占位符
127.0.0.1:6379> publish com. "hello3" 
(integer) 1

        (2) 具体实现

        在redisServer中有一个pubsub_patterns属性,该属性表示一个链表,链表中保存着所有和模式相关的信息。

struct redisServer {
    //...
    list *pubsub_patterns; 
    // ...
}

        pubsub_patterns链表的每个节点都包含一个 redis.h/pubsubPattern 结构:

typedef struct pubsubPattern {
    client *client;  -- 订阅模式客户端
    robj *pattern;   -- 被订阅的模式
} pubsubPattern;

        最终实现即如下图所示

         当有新的模式订阅者时,就将这个客户端分别加入到该模式的client中。

补充

        退订

        如要退订某个频道或模式,可使用如下指令:

  • UNSUBSCRIBE    UNSUBSCRIBE [channel [channel ...]]—取消订阅指定频道。
  • PUNSUBSCRIBE    PUNSUBSCRIBE [pattern [pattern ...]]—取消订阅符合指定模式的频道。

        发布订阅实际应用(待补充)

        Redis Sentinel 节点发现
        Redission 分布式锁​​​​​​​

        

        

Logo

华为开发者空间,是为全球开发者打造的专属开发空间,汇聚了华为优质开发资源及工具,致力于让每一位开发者拥有一台云主机,基于华为根生态开发、创新。

更多推荐