首页 > 数据库> 文章详细

Redis系列7-订阅发布机制

2021-09-09 23:01:50 阅读：136 来源： 互联网

标签：订阅频道系列 Redis channels client channel pubsub

订阅模式

Redis 提供两个订阅模式：频道（channel）订阅和 glob-style 模式（pattern）频道订阅。

频道订阅: CA（client A）向服务器订阅了频道 news，当 CB 向 news 发布消息的时候，CA 便能收到
glob-style频道订阅：举例，CA（client A）向服务器订阅了频道*.news：当 CB 向 cc.news、aa.news 发布消息的时候，CA 都能收到

数据结构

struct redisServer 和 struct redisClient 都维护了频道和模式频道，前者维护了所有频道和订阅频道的客户端，后者维护了客户端自己订阅的频道。

struct redisServer {
   ......
   /* Pubsub */
   dict *pubsub_channels; /* Map channels to list of subscribed clients */
   list *pubsub_patterns; /* A list of pubsub_patterns */
   ......
}
typedef struct redisClient {
   ......
   // 用户感兴趣的频道
   dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */
   // 用户感兴趣的模式
   list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */
   ......
} redisClient;
   // 模式频道数据结构，list *pubsub_patterns 里的每个节点数据都是struct
   // pubsubPattern。
typedef struct pubsubPattern {
   redisClient *client;
   robj *pattern;
} pubsubPattern;

频道订阅是一个 dict，每个 channel 被哈希进相应的桶，每个 channel 对应一个 clients，clients 都订阅了此 channel。当有消息发布的时候，检索 channel，遍历 clients，发布消息。
模式频道订阅是一个 list。当有消息发布的时候，channel 与 glob-style pattern 匹配，发布消息。

订阅过程

两种订阅模式是维护上述两种数据结构的过程，

// 订阅频道
/* 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(redisClient *c, robj *channel) {
   struct dictEntry *de;
   list *clients = NULL;
   int retval = 0;
   // redisClient.pubsub_channels 中保存客户端订阅的所有频道，可以查看客户端
   // 订阅了多少频道以及客户端是否订阅某个频道
   // server.pubsub_channels 中保存所有的频道和每个频道的订阅客户端，可以将
   // 消息发布到订阅客户端
   // 将频道加入redisClient.pubsub_channels
   /* Add the channel to the client -> channels hash table */
   if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
       retval = 1;
       incrRefCount(channel);
       // 在服务器负责维护的channel->clients 哈希表中寻找指定的频道
       /* Add the client to the channel -> list of clients hash table */
       de = dictFind(server.pubsub_channels,channel);
       // 未找到客户端指定的频道，需要创建
   if (de == NULL) {
       clients = listCreate();
       // 将频道加入server.pubsub_channels
       dictAdd(server.pubsub_channels,channel,clients);
       incrRefCount(channel);
       // 找到客户端指定的频道，直接获取这个频道
   } else {
       clients = dictGetVal(de);
   }
       // 将客户端添加到链表的尾部
       listAddNodeTail(clients,c);
   }
   // 通知客户端
   /* Notify the client */
   addReply(c,shared.mbulkhdr[3]);
   addReply(c,shared.subscribebulk);
   addReplyBulk(c,channel);
   addReplyLongLong(c,dictSize(c->pubsub_channels)+listLength(
   c->pubsub_patterns));
   return retval;
}
// 订阅模式频道
/* 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(redisClient *c, robj *pattern) {
   int retval = 0;
   // redisClient.pubsub_patterns 中保存客户端订阅的所有模式频道，可以查看
   // 客户端订阅了多少频道以及客户端是否订阅某个频道
   // server.pubsub_patterns 中保存所有的模式频道和每个模式频道的订阅客户端
   // ，可以将消息发布到订阅客户端
   // 未订阅模式频道，插入
   if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
       retval = 1;
       pubsubPattern *pat;
       // 将模式频道加入redisClient.pubsub_patterns
       listAddNodeTail(c->pubsub_patterns,pattern);
       incrRefCount(pattern);
       // 将模式频道加入server.pubsub_patterns
       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,dictSize(c->pubsub_channels)+listLength(
   c->pubsub_patterns));
   return retval;
}

消息发布

发布消息的过程则遍历上述两个数据结构（dict 和list），并将消息发布到匹配频道的所有客户端。

// 发布消息
/* Publish a message */
int pubsubPublishMessage(robj *channel, robj *message) {
   int receivers = 0;
   struct dictEntry *de;
   listNode *ln;
   listIter li;
   // 发布消息有两个步骤，
   // 指定频道的所有订阅者发布消息
   // 指定模式频道的所有订阅者发布消息
   //
   // 寻找频道
   /* Send to clients listening for that 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) {
       redisClient *c = ln->value;
       addReply(c,shared.mbulkhdr[3]);
       addReply(c,shared.messagebulk);
       addReplyBulk(c,channel);
       addReplyBulk(c,message);
       receivers++;
   }
}
//
// 进行glob-style 模式匹配
/* Send to clients listening to matching channels */
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;
}

注意，只要客户端订阅了频道，除了SUBCRIBE,UNSUBCRIBE,PSUBCRIBE,PSUBCRIBE，就不能执行其他命令。

int processCommand(redisClient *c) {
   ......
   // 在订阅发布模式下，只允许处理SUBSCRIBE 或者UNSUBSCRIBE 命令
   // 从下面的检测条件可以看出：只要存在redisClient.pubsub_channels 或者
   // redisClient.pubsub_patterns，就代表处于订阅发布模式下
   /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */
   if ((dictSize(c->pubsub_channels) > 0 || listLength(c->pubsub_patterns) > 0)
       &&
       c->cmd->proc != subscribeCommand &&
       c->cmd->proc != unsubscribeCommand &&
       c->cmd->proc != psubscribeCommand &&
       c->cmd->proc != punsubscribeCommand) {
       addReplyError(c,"only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed
               "in this context");
       return REDIS_OK;
       }
   ......
}

标签：订阅,频道,系列,Redis,channels,client,channel,pubsub
来源： https://www.cnblogs.com/mingpingyao/p/15249378.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

Redis系列7-订阅发布机制

订阅模式

数据结构

订阅过程

消息发布