谈到发布订阅模式, 相信不会陌生, 典型的观察者模式的实现. 然而从表面来看, 本地实现一个 wait/notify 通知, register/update 调用, 实现一个远程 mq 服务, 还有本文说的 pub/sub, 其实道理都差不多. 只是, 同样的需求, 针对不同的环境, 实现上往往是有天壤之别的.
所以, 我们就来看看 Redis 的 pub/sub 是如何实现的吧!
零, Redis 发布订阅相关概念介绍
Redis 发布订阅 (pub/sub) 是一种消息通信模式: 发送者 (pub) 发送消息, 订阅者 (sub) 接收消息. Redis 客户端可以订阅任意数量的频道.
下图展示了频道 channel1, 以及订阅这个频道的三个客户端 -- client2 , client5 和 client1 之间的关系:
当有新消息通过 PUBLISH 命令发送给频道 channel1 时, 这个消息就会被发送给订阅它的三个客户端:
Redis 的 pub/sub 实现中, 发布消息的方式只有一种, 但是订阅消息却有很多种方式.
使用场景如: 可以用做简单消息通信中间件, 监听某些事件的变化;
从官方手册上查到相关使用方法.
1> PUBLISH channel message
功能: 将信息发送到指定的频道.
返回值: 接收到该消息的个数;
2> SUBSCRIBE channel [channel ...]
功能: 订阅给定的一个或多个频道的信息.
返回值: 等待消息状态, 客户端不能再处理其他命令了. 除了 SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, PUNSUBSCRIBE, PING and QUIT commands.
3> PSUBSCRIBE pattern [pattern ...]
功能: 订阅一个或多个符合给定模式的频道.
返回值: 等待消息状态, 客户端不能再处理其他命令了. 除了 SUBSCRIBE, PSUBSCRIBE, UNSUBSCRIBE, PUNSUBSCRIBE, PING and QUIT commands.
4> PUBSUB subcommand [argument [argument ...]]
功能: 查看订阅与发布系统状态. subcomand 有 CHANNELS,NUMSUB,NUMPAT .
返回值:
PUBSUB CHANNELS [pattern] 列举出所有至少有一个订阅者的符合表达式的 channel(精确订阅的客户端, 即使用 SUBSCRIBE 进行订阅的客户端);
PUBSUB NUMSUB [channel-1 ... channel-N] 每个要查询的 channel 的订阅数 kv(精确订阅的客户端, 即使用 SUBSCRIBE 进行订阅的客户端);
PUBSUB NUMPAT 返回使用了 PSUBSCRIBE 订阅的客户端总数;
5> UNSUBSCRIBE [channel [channel ...]]
功能: 指退订给定的频道.
返回值: 退订频道的影响订阅数, 自身未订阅时, 影响数为 0;
6> PUNSUBSCRIBE [pattern [pattern ...]]
功能: 退订所有给定模式的频道.
返回值: 退订频道的影响订阅数, 自身未订阅时, 影响数为 0;
以上命令的操作, 当使用 Redis-cli 时, 将受限. 使用 SUBSCRIBE/PSUBSCRIBE 订阅 channel 后, 只能强行退出, 不能再接受其他命令. 即只有配合各语言实现的 sdk, 才能连贯完成上面完整的操作.
一, pub/sub 相关数据结构
pub/sub 相关接口定义如下:
- {"subscribe",subscribeCommand,-2,"rpslt",0,NULL,0,0,0,0,0},
- {"unsubscribe",unsubscribeCommand,-1,"rpslt",0,NULL,0,0,0,0,0},
- {"psubscribe",psubscribeCommand,-2,"rpslt",0,NULL,0,0,0,0,0},
- {"punsubscribe",punsubscribeCommand,-1,"rpslt",0,NULL,0,0,0,0,0},
- {"publish",publishCommand,3,"pltrF",0,NULL,0,0,0,0,0},
- {"pubsub",pubsubCommand,-2,"pltrR",0,NULL,0,0,0,0,0},
整个 pub/sub 使用的数据结构, 都是之前介绍过的. 主要有 dict, list 两种, 针对模式匹配订阅稍微多了个属性:
- // 使用 PSUBSCRIBE 订阅方式, 做一层数据格式封装
- typedef struct pubsubPattern {
- client *client;
- robj *pattern;
- } pubsubPattern;
二, subscribe/psubscribe 订阅 channel 实现
只有先有订阅者之后, 发布者发送的消息才会有意义. 所以我们先看看订阅的实现:
- // 用法: SUBSCRIBE channel [channel ...]
- // pubsub.c
- void subscribeCommand(client *c) {
- int j;
- // n 个 channel 的订阅, 循环调用即可
- for (j = 1; j <c->argc; j++)
- pubsubSubscribeChannel(c,c->argv[j]);
- // 添加 pubsub 订阅标识, 方便其他地方判断
- c->flags |= CLIENT_PUBSUB;
- }
- // 具体的单个 channel 订阅实现
- /* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
- * 0 if the client was already subscribed to that channel. */
- int pubsubSubscribeChannel(client *c, robj *channel) {
- dictEntry *de;
- list *clients = NULL;
- int retval = 0;
- /* Add the channel to the client -> channels hash table */
- // step1. 将要订阅的 channel 添加到各自客户端的 pubsub_channels 容器中
- if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
- retval = 1;
- incrRefCount(channel);
- /* Add the client to the channel -> list of clients hash table */
- // step2. 将要订阅的 channel 添加到 server.pubsub_channels 中, 方便在 publish 时判定是否触发通知
- de = dictFind(server.pubsub_channels,channel);
- if (de == NULL) {
- clients = listCreate();
- dictAdd(server.pubsub_channels,channel,clients);
- incrRefCount(channel);
- } else {
- clients = dictGetVal(de);
- }
- // step3. 将客户端自身添加到相应的 server.pubsub_channels 对应的队列中去, 在通知时只需遍历该队列即可
- listAddNodeTail(clients,c);
- }
- /* Notify the client */
- // 响应客户端:
- //*3 \r\n
- // $9\r\nsubscribe\r\n
- // channel
- // 111(该客户端总共订阅的 channel 数)
- addReply(c,shared.mbulkhdr[3]);
- addReply(c,shared.subscribebulk);
- addReplyBulk(c,channel);
- addReplyLongLong(c,clientSubscriptionsCount(c));
- return retval;
- }
- // 客户端订阅的总 channel 数, 两种订阅方式相加
- /* Return the number of channels + patterns a client is subscribed to. */
- int clientSubscriptionsCount(client *c) {
- return dictSize(c->pubsub_channels)+
- listLength(c->pubsub_patterns);
- }
如上就是单个 channel 的订阅方式了, 总结如下:
1. 客户端自行管理需要订阅的 channel, 放到 c->pubsub_channels 中;
2. Redis 使用的一个统一的 server->pubsub_channels dict 容器进行管理所有的 channel;
3. 对于多个客户端订阅一个 channel, Redis 使用 list 进行管理追加;
整个订阅过程, 其实就是一个注册的过程, 自然复杂不到哪里去. 接下来, 我们同步来看一下 使用模式订阅的方式的注册如何?
- // 用法: PSUBSCRIBE pattern [pattern ...]
- // pubsub.c
- void psubscribeCommand(client *c) {
- int j;
- // 同样是 n 个 channel 依次注册
- for (j = 1; j <c->argc; j++)
- pubsubSubscribePattern(c,c->argv[j]);
- c->flags |= CLIENT_PUBSUB;
- }
- // 注册单个模式匹配的 channel 订阅
- /* Subscribe a client to a pattern. Returns 1 if the operation succeeded, or 0 if the client was already subscribed to that pattern. */
- int pubsubSubscribePattern(client *c, robj *pattern) {
- int retval = 0;
- // 直接查找对应的 pattern, 没有则添加
- if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
- retval = 1;
- pubsubPattern *pat;
- listAddNodeTail(c->pubsub_patterns,pattern);
- incrRefCount(pattern);
- pat = zmalloc(sizeof(*pat));
- pat->pattern = getDecodedObject(pattern);
- pat->client = c;
- listAddNodeTail(server.pubsub_patterns,pat);
- }
- /* Notify the client */
- addReply(c,shared.mbulkhdr[3]);
- addReply(c,shared.psubscribebulk);
- addReplyBulk(c,pattern);
- addReplyLongLong(c,clientSubscriptionsCount(c));
- return retval;
- }
PSUBSCRIBE 的管理方式与 SUBSCRIBE 的管理方式不一样, 它是直接使用 list 保存订阅的模式到 server.pubsub_patterns 中, 针对不一样的模式, 使用一个新的 pubsubPattern 来保存.
注意: 所有客户端的订阅管理, server.pubsub_patterns 使用平坦式管理, 即相同的模式订阅, 有多少个客户端, 就会有多个元素被添加到 pubsub_patterns 中.(为什么不使用子链表的方式进行管理呢???)
三, publish 发布消息的实现
publish 是触发 subscribe 的方式, 没有 publish 动作, subscribe 就会一直在等待中. 想来应该不难, 消息发布之后, 只要将注册上来的客户一个进行消息推送, 就实现了相应功能. 所以, pub/sub 操作, 必然是基于长连接的实现方式, 没毛病.
Redis 的发布命令如下:
- // 用法: PUBLISH channel message
- // pubsub.c
- void publishCommand(client *c) {
- // 使用 channel+message 进行发布消息
- int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
- // 命令传播
- if (server.cluster_enabled)
- clusterPropagatePublish(c->argv[1],c->argv[2]);
- else
- forceCommandPropagation(c,PROPAGATE_REPL);
- addReplyLongLong(c,receivers);
- }
- // 发布一条消息
- /* Publish a message */
- int pubsubPublishMessage(robj *channel, robj *message) {
- int receivers = 0;
- dictEntry *de;
- listNode *ln;
- listIter li;
- /* Send to clients listening for that channel */
- // 使用 SUBSCRIBE 订阅的客户端, 直接遍历相应的 channel 集合即可
- de = dictFind(server.pubsub_channels,channel);
- if (de) {
- list *list = dictGetVal(de);
- listNode *ln;
- listIter li;
- // 依次进行数据响应, 将消息传播到订阅端
- listRewind(list,&li);
- while ((ln = listNext(&li)) != NULL) {
- client *c = ln->value;
- addReply(c,shared.mbulkhdr[3]);
- addReply(c,shared.messagebulk);
- addReplyBulk(c,channel);
- addReplyBulk(c,message);
- receivers++;
- }
- }
- /* Send to clients listening to matching channels */
- // 处理使用 PSUBSCRIBE 订阅消息的客户端
- // 前面说过, PSUBSCRIBE 在 Redis 使用平坦式管理, 所以需要做的模式匹配将会更多
- // 也就是说 PSUBSCRIBE 的响应性能也会更差
- if (listLength(server.pubsub_patterns)) {
- listRewind(server.pubsub_patterns,&li);
- channel = getDecodedObject(channel);
- while ((ln = listNext(&li)) != NULL) {
- pubsubPattern *pat = ln->value;
- if (stringmatchlen((char*)pat->pattern->ptr,
- sdslen(pat->pattern->ptr),
- (char*)channel->ptr,
- sdslen(channel->ptr),0)) {
- addReply(pat->client,shared.mbulkhdr[4]);
- addReply(pat->client,shared.pmessagebulk);
- addReplyBulk(pat->client,pat->pattern);
- addReplyBulk(pat->client,channel);
- addReplyBulk(pat->client,message);
- receivers++;
- }
- }
- decrRefCount(channel);
- }
- return receivers;
- }
可以看到, Redis 消息的发布可能比想象中的还要简单. 不过有一点需要注意的是, 整个 publish 的消息并没有在 Redis 进行存储操作, 也就是说发布完一个消息之后, 就再也找不到踪迹了. 这也是很多消息中间件的实现方式, 因为数据的保留可能会显得没有意义.
整个发布消息的过程, 其实就是向各个 subscriber 进行数据推送的过程, 而这些 scriber 则是基于长连接客户端实例, 以至于其看起来和本地实现的 register/update 的观察者模块没啥两样.
所以, 基本上 Redis 的发布订阅功能实现得, 还是实现的粒度还是比较粗的. 系统上的应用如哨兵模式下的 master/slave 的切换. 而如果自己应用的话, 就需要找准自己的应用场景, 不要乱用了.
四, unsubscribe 解除订阅关系
当关注的事件处理完成后, 可能就不需要再订阅相关消息了, 就需要进行解决订阅. 解决订阅关系, 即不再接受相应的发布消息, 将自身从注册表中删除即可. 基本上就是和订阅进行一个反解操作!
- // 用法: UNSUBSCRIBE [channel [channel ...]]
- // pubsub.c
- void unsubscribeCommand(client *c) {
- // unsubscribe, 直接解决所有的订阅
- if (c->argc == 1) {
- pubsubUnsubscribeAllChannels(c,1);
- } else {
- int j;
- // 根据指定的 channel 依次解除订阅关系
- for (j = 1; j <c->argc; j++)
- pubsubUnsubscribeChannel(c,c->argv[j],1);
- }
- // 当一个订阅也没有, 则自身不再处理 pubsub 相关的事务
- if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
- }
- /* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
- * 0 if the client was not subscribed to the specified channel. */
- int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
- dictEntry *de;
- list *clients;
- listNode *ln;
- int retval = 0;
- /* Remove the channel from the client -> channels hash table */
- incrRefCount(channel); /* channel may be just a pointer to the same object
- we have in the hash tables. Protect it... */
- // 先删除自身的订阅标识, 再删除 server.pubsub_channels 标识
- if (dictDelete(c->pubsub_channels,channel) == DICT_OK) {
- retval = 1;
- /* Remove the client from the channel -> clients list hash table */
- de = dictFind(server.pubsub_channels,channel);
- serverAssertWithInfo(c,NULL,de != NULL);
- clients = dictGetVal(de);
- ln = listSearchKey(clients,c);
- serverAssertWithInfo(c,NULL,ln != NULL);
- listDelNode(clients,ln);
- if (listLength(clients) == 0) {
- /* Free the list and associated hash entry at all if this was
- * the latest client, so that it will be possible to abuse
- * Redis PUBSUB creating millions of channels. */
- dictDelete(server.pubsub_channels,channel);
- }
- }
- /* Notify the client */
- // 调用 unsubscribe 进行解决订阅的, 此处都需要进行客户端响应通知
- if (notify) {
- addReply(c,shared.mbulkhdr[3]);
- addReply(c,shared.unsubscribebulk);
- addReplyBulk(c,channel);
- addReplyLongLong(c,dictSize(c->pubsub_channels)+
- listLength(c->pubsub_patterns));
- }
- decrRefCount(channel); /* it is finally safe to release it */
- return retval;
- }
- // 解决所有的当前客户端的订阅关系 (SUBSCRIBE 建立的订阅)
- /* Unsubscribe from all the channels. Return the number of channels the
- * client was subscribed to. */
- int pubsubUnsubscribeAllChannels(client *c, int notify) {
- dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
- dictEntry *de;
- int count = 0;
- // 迭代 c->pubsub_channels 的订阅, 依次删除即可
- while((de = dictNext(di)) != NULL) {
- robj *channel = dictGetKey(de);
- count += pubsubUnsubscribeChannel(c,channel,notify);
- }
- /* We were subscribed to nothing? Still reply to the client. */
- if (notify && count == 0) {
- addReply(c,shared.mbulkhdr[3]);
- addReply(c,shared.unsubscribebulk);
- addReply(c,shared.nullbulk);
- addReplyLongLong(c,dictSize(c->pubsub_channels)+
- listLength(c->pubsub_patterns));
- }
- dictReleaseIterator(di);
- return count;
- }
不出所料, 就是一个从 pubsub_channels 中的删除一个元素的问题, 别无其他. 其中需要注意的是, SUBSCRIBE 对应 UNSUBSCRIBE, PSUBSCRIBE 对应 PUNSUBSCRIBE.
四, 关于 Redis pub/sub 之后的思考
需要注意的是, 消息中间件是远程通信组件, 必然存在各种不确定性, 所以确保长连接的有效性是非常重要, 比如通过 PING-PONG 方式进行续租, 以保持连接的有效性.
可以说, 我们要实现一个简单的 pub/sub 功能是简单的, 但是要应对各种异常情况则是困难的.
1. 比如当订阅的量越来越大时, 整个发布消息过程可能变量缓慢起来, 如何处理?
2. 如果消费者端处理失败, 如何处理?
3. 订阅者为什么只能做很少的事情, 能不能多做一点?
4. 出现问题时如何进行溯源?
5. 如何处理单机瓶颈问题?
6. 如果是多机负载, 如何处理数据一致性问题?
7. 消费者事务处理能力问题?
Redis 是专业的缓存解决方案, 但不是专业的消息通信解决方案, 它的实现只能为打开我们的一点思路. 我们还是要相信专业的力量, 以上问题相信在很多消息中间件中很容易找到相应答案.
来源: https://www.cnblogs.com/yougewe/p/12349899.html