Redis - PubSub(发布者-订阅者模式)的内部实现

一.设计模式-发布订阅模式

发布订阅模式，又叫观察者模式，属于四人帮的二十三个设计模式中的行为模式。”定义对象间一种一对多的依赖关系，使得每当一个对象改变状态，则所有依赖于他的对象都会得到通知并被自动更新“，模式UML如下图。

通俗一点可以理解为，Subject中保存了Observer的引用组成的列表。Subject状态变化时，遍历列表调用所有Observer的notify成员方法。
发布订阅模式应用场景很多，相当于在系统中构建一个简单的事件处理系统，部分编程语言（如java提供被观察者类java.util.Observable和观察者接口java.util.Observer）原生库中已包含观察者模式的简单实现。

二.Redis-PubSub介绍

Redis中的PubSub实现较上面复杂，但设计思想和上面一致。

客户端A发出一个订阅的频道名称(可以理解为attach A 到Subject)

subscribe  temperature  rainfall

其他客户端B发布消息（可以理解为B更新Subjec的状态）

publish  rainfall 350cc

A客户端将收到降雨量350cc这条message

客户端A也可以取消订阅频道（可以理解为A从Subjec中detach）

unsubscribe rainfall

在提供订阅频道外，Redis还提供了psubscribe/punsubscribe来订阅/取消订阅一个或多个模式。如果模式照通配符(glob)来匹配某个/ 某些频道，当有信息发送给这个/这些频道的时候， 客户端也会收到这个/这些频道的信息。

二.Redis-PubSub订阅频道

Redis中PubSub实现也比较简单，订阅频道的功能主要保存在字典中，其中key为频道，而value是redisClient组成的列表。

下图为订阅频道的redisServer的pubsub_channels字典示例，ClientA订阅 temperature和rainfall，ClientC订阅temperature



当客户端ClientD订阅已经存在的频道时，将ClientD加到该频道对应的链表的尾巴，如订阅rainfall频道。
当客户端ClientD订阅不存在的频道时，新建新频道的key及对应的链表，将ClientD加到新创建的链表中，下图为ClientD执行 subscribe snow后的字典示例。



当客户端ClientA取消订阅频道时，将客户端ClientA从对应的频道的链表中删除，如果删除后对应的频道无客户订阅，则将对应的频道的key从字典中删除。
如ClientA执行subscribe（无参数的subscribe 命令订阅的所有频道都会被退订）后字典的示例。



以上的所有操作在更新redisServer的pubsub_channels字典的同时也会更新redisClient中的pubsub_channels字典，redisClient的pubsub_channels保存着对应客户端订阅的频道。


四.Redis-PubSub订阅模式

订阅模式的功能保存在redisServer的pubsub_patterns链表中，链表的元素结构如下：

typedef struct pubsubPattern {
   /*客户端信息，唯一确认一个客户连接*/
    redisClient *client;
   /*模式信息的字符串对象，如"*no*"*/
    robj *pattern;
} pubsubPattern;

下图示例客户端ClientA 订阅“*it*”、"*eye*"，客户端ClientB订阅"*eye*"的链表。



增加时遍历redisClient的pubsub_patterns链表，如果不存在，将新的模式增加到redisServer的pubsub_patterns链表的尾部。

删除时需要遍历redisClient的整个pubsub_patterns，如果存在，在定位redisServer的pubsub_patterns链表进行删除操作。

与订阅频道同样，在更新redisServer的pubsub_patterns链表的同时，也更新redisClient的pubsub_patterns链表，redisClient的pubsub_channels保存着对应客户端订阅的模式。

五.消息发送publish

当发送消息时，先从redisServer找到对应的频道的key所拥有的链表，遍历链表发送消息；然后遍历redisServer的pubsub_patterns链表，对频道和pattern进行匹配（stringmatch），如果适配将信息增加到客户端的返回中。
发送消息的伪代码如下：

/*
* 发送消息
* channel 频道
* message  消息主体
* */
function pubsubPublishMessage(channel,message){

    /*需要channel被某些客户端订阅，如果不存在pubsub_channels中，就没有订阅此频道的客户*/
    if (isset(server.pubsub_channels[channel])){
        /*
         * 遍历pubsub_channels中对应频道的订阅者的链表
         * 给每个频道订阅者发布消息
         * */
        foreach (server.pubsub_channels[channel] as receiver){

            sendMessage(receiver, message);
        }

    }
    /*遍历模式订阅的链表*/
   foreach (server . pubsub_patterns as pubsubPattern){

        /*如果模式与频道适配，发送消息给此客户端*/
        if (stringmatch(pubsubPattern.pattern, channel)) {

            sendMessage(pubsubPattern.client, message);

        }

    }

}

从上面代码中可以Redis发布的消息“即发即失”，redis不会持久保存发布的消息， 一旦client端断开链接，将会失去部分消息。
如果需要client非常关注的消息（如用户支付订单成功，client发货），建议不要用Redis的PubSub功能，可以用其他专门的消息队列系统，也可用Redis的链表，然后客户端进行轮询。