文章目录

• 前言
• 一、牛刀小试
• • 1.订阅
  • 2.模式订阅：
• 二、原理
• • 1.服务端
  • 2.客户端
  • 3.应用场景
• 总结

---

前言

本文参考源码版本为 redis6.2，redisson3.17.5

发布订阅模式，本质来说，是将提供消息的人和需要消息的人，通过第三方组件联系起来，使得两类群体之间的消息能够及时触达。

比如，在一些优化场景下，可能会使用 本地 + 远程 双缓存机制，远程缓存是一套共用的中间件，总共只有一套数据。而 本地缓存就不一样了，如果你部署的是多个实例，那就有多套本地数据，当数据更新了，如何触达这些本地缓存？

这个时候，你就可以考虑使用发布订阅模式，消息提供者 - 更新数据的人，消息接收方 - 需要更新本地缓存的服务。

我们以 redis 发布订阅的实现来看，其实是非常简单的，从原理到实际编码都是如此。

• redis 维护 通道与订阅者的关系
• 发布者将消息推送至对应通道
• redis 将消息转发至与该通道关联的所有订阅者

值得注意的是，上述消息不会持久化，都是直来直去，立即转发。因此，你会看到，如果某个订阅者掉线了，后面即使上线了，这掉线期间的消息也无法重新接收到。

诶，你发现，服务端实现是不是很简单？基本上维护服务端与订阅者的长连接即可，后续有消息过来直接推送；当订阅者取消订阅或者掉线，服务端删除对应订阅关系即可。

好，说完了服务端，我们来看看客户端该怎么做？

客户端的订阅者想要接收消息，肯定也会维护长连接，因此，当我们订阅通道时，本质就是向 redis 服务发送订阅请求，并维护此长连接，用于后续消息通信。

一、牛刀小试

1.订阅

SUBSCRIBE channels

当发起订阅之后，redis-cli 进入阻塞状态，这个时候无法发送其他命令，也就是处于监听消息的状态：

127.0.0.1:6379> SUBSCRIBE foo bar
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "foo"
3) (integer) 1
1) "subscribe"
2) "bar"
3) (integer) 2

然后，我们往 foo 通道发送消息：

127.0.0.1:6379> PUBLISH foo hello
(integer) 1

看看接收方：

127.0.0.1:6379> SUBSCRIBE foo bar
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "foo"
3) (integer) 1
1) "subscribe"
2) "bar"
3) (integer) 2
1) "message"
2) "foo"
3) "hello"

你会看到最后三条数据，message 表示类型、foo 表示通道、hello 表示具体消息。实时触达。

2.模式订阅：

你可能也注意到了，这里我们可以一次性订阅多个通道。当你订阅几个相似的通道时，比如这样：

127.0.0.1:6379> SUBSCRIBE foo.a foo.b foo.c foo.d
Reading messages... (press Ctrl-C to quit)
1) "subscribe"
2) "foo.a"
3) (integer) 1
1) "subscribe"
2) "foo.b"
3) (integer) 2
1) "subscribe"
2) "foo.c"
3) (integer) 3
1) "subscribe"
2) "foo.d"
3) (integer) 4

然后就可以接收任何一个通道发过来的信息。

你看，这四个通道都有相同的前缀 foo.，redis 也考虑到了这点，提供了模式订阅方式 PSUBSCRIBE：

127.0.0.1:6379> PSUBSCRIBE foo.*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "foo.*"
3) (integer) 1

往四个通道发送数据看看：

127.0.0.1:6379> PUBLISH foo.a hello_a
(integer) 1
127.0.0.1:6379> PUBLISH foo.b hello_b
(integer) 1
127.0.0.1:6379> PUBLISH foo.c hello_c
(integer) 1
127.0.0.1:6379> PUBLISH foo.d hello_d
(integer) 1

再看看接收方：

127.0.0.1:6379> PSUBSCRIBE foo.*
Reading messages... (press Ctrl-C to quit)
1) "psubscribe"
2) "foo.*"
3) (integer) 1
1) "pmessage"
2) "foo.*"
3) "foo.a"
4) "hello_a"
1) "pmessage"
2) "foo.*"
3) "foo.b"
4) "hello_b"
1) "pmessage"
2) "foo.*"
3) "foo.c"
4) "hello_c"
1) "pmessage"
2) "foo.*"
3) "foo.d"
4) "hello_d"

可以看到，4 个通道的消息都能接收到，其实不止，只要匹配这个模式 foo. 消息都能够接收到；另外，消息类型也变成了 pmessage

二、原理

PubSub 也是一种生产者/消费者模型，和传统 MQ 不同的是，经过 PubSub 的消息只做转发，不做存储，从实现上来更加简单。

我画了张图，大概是这样：

不知你有没有思考过，服务端是如何推送消息给对应的订阅者的？怎么来识别客户端？IP + PORT 还是什么唯一的身份标志？

其实，redis 处理方式很简单，只是一直维护着这些订阅的 TCP 长连接请求。你想想，这些连接都是经过 TCP 三次握手建立的，只要双方都没断开，便可一直保持通信。这样一来，服务端要做的事情就很简单了是吧？

在服务端看来，客户端发来的 subscribe 指令与普通请求并无差别，服务端在接收到请求时，会加一个 flag 标志，之后便将这些连接保存在特定的结构中，等待 publish 指令的到来。

当服务端收到 publish 指令后，会匹配相应的 channel，并将消息直接推送给订阅的客户端。所以，你会看到，不管是客户端还是服务端，都需要维护这么一个长连接。

另外，对于客户端来说，也同样没有什么特别之处，创建 TCP 连接之后，将指令往连接通道一发就完事了。

不过，由于这些订阅连接需要一直保持状态以监听服务端的消息到来，因此，在客户端设计中，通常会把常规指令和 PubSub 指令分开处理，比如 Redisson，就分别定义了不同的连接池。

总的来说，客户端与服务端是通过 TCP 长连接持续通信，一个客户端实例可以创建很多个这样的长连接，因此，客户端的设计中，一般也会给每个客户端实例设置一个专用连接池。

从 PubSub 这一块内容来看，要想设计一个款高效的客户端，客户端的工作也不少~

1.服务端

我们先看看对 Pubsub 的结构定义：

struct redisServer {
     
    ...
     
    /* Pubsub */
    dict *pubsub_channels;  /* Map channels to list of subscribed clients */
    list *pubsub_patterns;  /* A list of pubsub_patterns */
    dict *pubsub_patterns_dict;  /* A dict of pubsub_patterns */
    
    ...
    
}    

• pubsub_channels：通道与客户端的订阅关系，是一个字典，key 是 channel，value 是订阅的客户端链表
• pubsub_patterns：模式匹配，记录的是模式订阅的客户端列表
• pubsub_patterns_dict：模式匹配字典，后期加上的，主要考虑模式订阅的客户端过多之后的效率问题，索性就直接搞成 pubsub_channels 类似的字典，模式通道为 key，value 是订阅的客户端列表

可以看到，服务端的结构中，只记录了通道与客户端订阅关系，没有任何消息会被保存下来。

我画了两张图，你可以参考下：

1）pubsub_channels 的字典结构：

2）pubsub_patterns 的链表模式：

当然，pubsub_patterns 后面也提供了字典结构 pubsub_patterns_dict。

订阅和取消订阅：本质都是在维护以上订阅关系列表，比如收到订阅请求时，会遍历 pubsub_channels 或者 pubsub_patterns 结构，然后将请求添加至链表尾部。

值得注意的是，这里的客户端，本质也是 TCP 连接，可随时互动通信。

消息发布：当服务端接收到发布的消息之后，会先遍历 pubsub_channels 字典，如果找到对应的订阅客户端，则直接推送消息，直到遍历完整个 pubsub_channels 字典。

然后遍历 pubsub_patterns 链表，对匹配上的订阅客户端，同样是直接推送消息。因此，你会看到，即使同样一条消息，匹配多个模式之后，也会推送多次。

2.客户端

试想一下，如果你要订阅一个 channel，你会如何定义客户端？

首先，要想订阅 channel 肯定需要向服务端发送 subscribe channels 指令。

其次，发送指令订阅了 channel 之后只是建立了正常连接，后续还需要监听服务端推送的消息，因此，需要定义监听方法；不过，对于同一个 channel 我们可能同时有多种处理方式，所以，这里的监听器需要定义成列表形式。

另外，我们可能在不同时刻需要和多个 channel 存在订阅关系，因此会存在多个长连接，此时，我们考虑使用连接池。

那连接池是考虑定义专用的还是和普通请求的连接池公用？考虑 PubSub 的特殊性，我们决定使用专用的 PubSub 连接池。

再来看看 Redisson 客户端，通常情况下，我们会先定义一个 RedissonClient 实例，这个 RedissonClient 客户端下就会有多种连接池，其中就包括 PubSub 的专用连接池 PubSubConnectionPool。

值得注意的是，subscribe 指令是阻塞式的，因此，一般考虑使用异步线程池来提交请求；在 redisson 中，使用 Netty 自带异步功能来完成，本质也是使用异步线程池。

以 redisson 客户端为例，我画了张图，大概是这样：

3.应用场景

PubSub 有 MQ 的一些特性，比如 解藕，另外还有消息及时转发等。

当你使用 服务本地内存 + redis 等中间缓存等双缓存特性时，在这种情况下你就需要考虑本地缓存更新的及时性，PubSub 便可以做到。

另外，我们在来看看 redisson，在分布式场景的多节点下，比如我们要同时获取一把锁，同一时间肯定只有一个节点能正常获取锁，其他节点只能等待锁的释放，那需要等待多久呢？

一般通过 while(true) 循环尝试获取锁，每循环一次都尝试休眠一段时间；假设在节点休眠期间锁释放了，该节点反应是不是就迟钝了？

类似于这样的情况，也可以考虑使用 PubSub，当节点处于等待获取锁的情况下，先订阅一个通道，等到锁释放之后，持有锁的节点发送通知，然后等待锁的节点基本上可以实时接收通知，并尝试获取锁。

总结

发布订阅也是一种生产者、消费者模型，借助于第三方组件，解藕两者关系，避免直接交互。

redis 实现的 PubSub 比较简单，消息在服务端只做转发，不做存储。

另外，服务端会维护订阅者的 TCP 连接，方便消息及时触达。

相关参考：

• Redis Pub/Sub