参考资料：

《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
# 返回结果：接收到信息的订阅者数量，无订阅者返回0127.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 分布式锁​​​​​​​