一文带你了解 Redis 的发布与订阅的底层原理（上）

01、前言

发布订阅系统在我们日常的工作中经常会使用到，这种场景大部分情况我们都是使用消息队列的，常用的消息队列有 Kafka，RocketMQ，RabbitMQ，每一种消息队列都有其特性，关于 Kafka 的使用和源码分析，公号前面有相关的文章，大家可以前往回顾一下，另外两款消息队列大家有需要可以自行研究，后续我们会出相应的介绍文章。这篇文章主要是给大家介绍 Redis 的发布订阅系统，很多时候我们可能不需要独立部署相应的消息队列，只是简单的使用，而且数据量也不会太大，这种情况下，我们就可以使用 Redis 的 Pub/Sub 模型。

02、使用方式

2.1 发布与订阅

Redis 的发布订阅功能主要由 PUBLISH，SUBSCRIBE，PSUBSCRIBE 命令组成，一个或者多个客户端订阅某个或者多个频道，当其他客户端向该频道发送消息的时候，订阅了该频道的客户端都会收到对应的消息。

上图中有四个客户端，Client 02，Client 03，Client 04 订阅了同一个 Sport 频道（Channel），这时当 Client 01 向 Sport Channel 发送消息 “basketball” 的时候，02-04 这三个客户端都同时收到了这条消息。

整个过程的执行命令如下：

首先开四个 Redis 的客户端，然后在 Client 02，Client 03，Client 04 中输入subscribe sport 命令，表示订阅 sport 这个频道

然后在 Client 01 的客户端中输入publish sport basketball 表示向 sport 频道发送消息 "basketball"

这个时候我们在去看下 Client 02-04 的客户端，可以看到已经收到了消息了，每个订阅了这个频道的客户端都是一样的。

这里 Client 02-Client 04 三个客户端订阅了 Sport 频道，我们叫做订阅者（subscriber），Client 01 发布消息，我们叫做发布者（publisher），发送的消息就是 message。

2.2、模式订阅

前面我们看到的是一个客户端订阅了一个 Channel，事实上单个客户端也可以同时订阅多个 Channel，采用模式匹配的方式，一个客户端可以同时订阅多个 Channel。

如上图 Client 05 通过命令subscribe run 订阅了 run 频道，Client 06 通过命令psubscribe run* 订阅了 run* 匹配的频道。当 Client 07 向 run 频道发送消息 666 的时候，05 和 06 两个客户端都收到消息了；接下来 Client 07 向 run1 和 run_sport 两个频道发送消息的时候，Client 06 依旧可以收到消息，而 Client 05 就收不到了消息了。

Client 05 订阅run 频道和接收到消息：

Client 06 订阅run* 频道和接收到消息：

Client 07 向多个频道发送消息：

通过上面的案例，我们学会了一个客户端可以订阅单个或者多个频道，分别通过subscribe，psubscribe 命令，客户端可以通过 publish 发送相应的消息。

在命令行中我们可以用 Ctrl + C 来取消相关订阅，对应的命令时 unsubscribe channelName。

03、Pub/Sub 底层存储结构

3.1、订阅 Channel

在 Redis 的底层结构中，客户端和频道的订阅关系是通过一个字典加链表的结构保存的，形式如下：

在 Redis 的底层结构中，Redis 服务器结构体中定义了一个 pubsub_channels 字典

struct redisServer {
  //用于保存所有频道的订阅关系
  dict *pubsub_channels;
}

在这个字典中，key 代表的是频道名称，value 是一个链表，这个链表里面存放的是所有订阅这个频道的客户端。

所以当有客户端执行订阅频道的动作的时候，服务器就会将客户端与被订阅的频道在 pubsub_channels 字典中进行关联。

这个时候有两种情况：

• 该渠道是首次被订阅：首次被订阅说明在字典中并不存在该渠道的信息，那么程序首先要创建一个对应的 key，并且要赋值一个空链表，然后将对应的客户端加入到链表中。此时链表只有一个元素。
• 该渠道已经被其他客户端订阅过：这个时候就直接将对应的客户端信息添加到链表的末尾就好了。

比如，如果有一个新的客户端 Client 08 要订阅 run 渠道，那么上图就会变成

如果 Client 08 要订阅一个新的渠道 new_sport ，那么就会变成

整个订阅的过程可以采用下面伪代码来实现

Map<String, List<Object>> pubsub_channels = new HashMap<>();
    public void subscribe(String[] subscribeList, Object client) {
        //遍历所有订阅的 channel，检查是否在 pubsub_channels 中，不在则创建新的 key 和空链表
        for (int i = 0; i < subscribeList.length; i++) {
            if (!pubsub_channels.containsKey(subscribeList[i])) {
                pubsub_channels.put(subscribeList[i], new ArrayList<>());
            }
            pubsub_channels.get(subscribeList[i]).add(client);
        }
    }

3.2 取消订阅

上面介绍的是单个 Channel 的订阅，相反的如果一个客户端要取消订阅相关 Channel，则无非是找到对应的 Channel 的链表，从中删除对应的客户端，如果该客户端已经是最后一个了，则将对应 Channel 也删除。

public void unSubscribe(String[] subscribeList, Object client) {
        //遍历所有订阅的 channel，依次删除
        for (int i = 0; i < subscribeList.length; i++) {
            pubsub_channels.get(subscribeList[i]).remove(client);
            //如果长度为 0 则清楚 channel
            if (pubsub_channels.get(subscribeList[i]).size() == 0) {
                remove(subscribeList[i]);
            }
        }
    }