关键词搜索

源码搜索 ×
×

漫话Redis源码之三十六

发布2021-12-12浏览1498次

详情内容

这里主要是Pubsub相关的API, 大致看看就行，知道他的功能就基本差不多了：

  1. #include "server.h"
  2.  
  3. int clientSubscriptionsCount(client *c);
  4.  
  5. /*-----------------------------------------------------------------------------
  6.  * Pubsub client replies API
  7.  *----------------------------------------------------------------------------*/
  8.  
  9. /* Send a pubsub message of type "message" to the client.
  10.  * Normally 'msg' is a Redis object containing the string to send as
  11.  * message. However if the caller sets 'msg' as NULL, it will be able
  12.  * to send a special message (for instance an Array type) by using the
  13.  * addReply*() API family. */
  14. void addReplyPubsubMessage(client *c, robj *channel, robj *msg) {
  15.     if (c->resp == 2)
  16.         addReply(c,shared.mbulkhdr[3]);
  17.     else
  18.         addReplyPushLen(c,3);
  19.     addReply(c,shared.messagebulk);
  20.     addReplyBulk(c,channel);
  21.     if (msg) addReplyBulk(c,msg);
  22. }
  23.  
  24. /* Send a pubsub message of type "pmessage" to the client. The difference
  25.  * with the "message" type delivered by addReplyPubsubMessage() is that
  26.  * this message format also includes the pattern that matched the message. */
  27. void addReplyPubsubPatMessage(client *c, robj *pat, robj *channel, robj *msg) {
  28.     if (c->resp == 2)
  29.         addReply(c,shared.mbulkhdr[4]);
  30.     else
  31.         addReplyPushLen(c,4);
  32.     addReply(c,shared.pmessagebulk);
  33.     addReplyBulk(c,pat);
  34.     addReplyBulk(c,channel);
  35.     addReplyBulk(c,msg);
  36. }
  37.  
  38. /* Send the pubsub subscription notification to the client. */
  39. void addReplyPubsubSubscribed(client *c, robj *channel) {
  40.     if (c->resp == 2)
  41.         addReply(c,shared.mbulkhdr[3]);
  42.     else
  43.         addReplyPushLen(c,3);
  44.     addReply(c,shared.subscribebulk);
  45.     addReplyBulk(c,channel);
  46.     addReplyLongLong(c,clientSubscriptionsCount(c));
  47. }
  48.  
  49. /* Send the pubsub unsubscription notification to the client.
  50.  * Channel can be NULL: this is useful when the client sends a mass
  51.  * unsubscribe command but there are no channels to unsubscribe from: we
  52.  * still send a notification. */
  53. void addReplyPubsubUnsubscribed(client *c, robj *channel) {
  54.     if (c->resp == 2)
  55.         addReply(c,shared.mbulkhdr[3]);
  56.     else
  57.         addReplyPushLen(c,3);
  58.     addReply(c,shared.unsubscribebulk);
  59.     if (channel)
  60.         addReplyBulk(c,channel);
  61.     else
  62.         addReplyNull(c);
  63.     addReplyLongLong(c,clientSubscriptionsCount(c));
  64. }
  65.  
  66. /* Send the pubsub pattern subscription notification to the client. */
  67. void addReplyPubsubPatSubscribed(client *c, robj *pattern) {
  68.     if (c->resp == 2)
  69.         addReply(c,shared.mbulkhdr[3]);
  70.     else
  71.         addReplyPushLen(c,3);
  72.     addReply(c,shared.psubscribebulk);
  73.     addReplyBulk(c,pattern);
  74.     addReplyLongLong(c,clientSubscriptionsCount(c));
  75. }
  76.  
  77. /* Send the pubsub pattern unsubscription notification to the client.
  78.  * Pattern can be NULL: this is useful when the client sends a mass
  79.  * punsubscribe command but there are no pattern to unsubscribe from: we
  80.  * still send a notification. */
  81. void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) {
  82.     if (c->resp == 2)
  83.         addReply(c,shared.mbulkhdr[3]);
  84.     else
  85.         addReplyPushLen(c,3);
  86.     addReply(c,shared.punsubscribebulk);
  87.     if (pattern)
  88.         addReplyBulk(c,pattern);
  89.     else
  90.         addReplyNull(c);
  91.     addReplyLongLong(c,clientSubscriptionsCount(c));
  92. }
  93.  
  94. /*-----------------------------------------------------------------------------
  95.  * Pubsub low level API
  96.  *----------------------------------------------------------------------------*/
  97.  
  98. /* Return the number of channels + patterns a client is subscribed to. */
  99. int clientSubscriptionsCount(client *c) {
  100.     return dictSize(c->pubsub_channels)+
  101.            listLength(c->pubsub_patterns);
  102. }
  103.  
  104. /* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
  105.  * 0 if the client was already subscribed to that channel. */
  106. int pubsubSubscribeChannel(client *c, robj *channel) {
  107.     dictEntry *de;
  108.     list *clients = NULL;
  109.     int retval = 0;
  110.  
  111.     /* Add the channel to the client -> channels hash table */
  112.     if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
  113.         retval = 1;
  114.         incrRefCount(channel);
  115.         /* Add the client to the channel -> list of clients hash table */
  116.         de = dictFind(server.pubsub_channels,channel);
  117.         if (de == NULL) {
  118.             clients = listCreate();
  119.             dictAdd(server.pubsub_channels,channel,clients);
  120.             incrRefCount(channel);
  121.         } else {
  122.             clients = dictGetVal(de);
  123.         }
  124.         listAddNodeTail(clients,c);
  125.     }
  126.     /* Notify the client */
  127.     addReplyPubsubSubscribed(c,channel);
  128.     return retval;
  129. }
  130.  
  131. /* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
  132.  * 0 if the client was not subscribed to the specified channel. */
  133. int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
  134.     dictEntry *de;
  135.     list *clients;
  136.     listNode *ln;
  137.     int retval = 0;
  138.  
  139.     /* Remove the channel from the client -> channels hash table */
  140.     incrRefCount(channel); /* channel may be just a pointer to the same object
  141.                             we have in the hash tables. Protect it... */
  142.     if (dictDelete(c->pubsub_channels,channel) == DICT_OK) {
  143.         retval = 1;
  144.         /* Remove the client from the channel -> clients list hash table */
  145.         de = dictFind(server.pubsub_channels,channel);
  146.         serverAssertWithInfo(c,NULL,de != NULL);
  147.         clients = dictGetVal(de);
  148.         ln = listSearchKey(clients,c);
  149.         serverAssertWithInfo(c,NULL,ln != NULL);
  150.         listDelNode(clients,ln);
  151.         if (listLength(clients) == 0) {
  152.             /* Free the list and associated hash entry at all if this was
  153.              * the latest client, so that it will be possible to abuse
  154.              * Redis PUBSUB creating millions of channels. */
  155.             dictDelete(server.pubsub_channels,channel);
  156.         }
  157.     }
  158.     /* Notify the client */
  159.     if (notify) addReplyPubsubUnsubscribed(c,channel);
  160.     decrRefCount(channel); /* it is finally safe to release it */
  161.     return retval;
  162. }
  163.  
  164. /* Subscribe a client to a pattern. Returns 1 if the operation succeeded, or 0 if the client was already subscribed to that pattern. */
  165. int pubsubSubscribePattern(client *c, robj *pattern) {
  166.     dictEntry *de;
  167.     list *clients;
  168.     int retval = 0;
  169.  
  170.     if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
  171.         retval = 1;
  172.         listAddNodeTail(c->pubsub_patterns,pattern);
  173.         incrRefCount(pattern);
  174.         /* Add the client to the pattern -> list of clients hash table */
  175.         de = dictFind(server.pubsub_patterns,pattern);
  176.         if (de == NULL) {
  177.             clients = listCreate();
  178.             dictAdd(server.pubsub_patterns,pattern,clients);
  179.             incrRefCount(pattern);
  180.         } else {
  181.             clients = dictGetVal(de);
  182.         }
  183.         listAddNodeTail(clients,c);
  184.     }
  185.     /* Notify the client */
  186.     addReplyPubsubPatSubscribed(c,pattern);
  187.     return retval;
  188. }
  189.  
  190. /* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
  191.  * 0 if the client was not subscribed to the specified channel. */
  192. int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
  193.     dictEntry *de;
  194.     list *clients;
  195.     listNode *ln;
  196.     int retval = 0;
  197.  
  198.     incrRefCount(pattern); /* Protect the object. May be the same we remove */
  199.     if ((ln = listSearchKey(c->pubsub_patterns,pattern)) != NULL) {
  200.         retval = 1;
  201.         listDelNode(c->pubsub_patterns,ln);
  202.         /* Remove the client from the pattern -> clients list hash table */
  203.         de = dictFind(server.pubsub_patterns,pattern);
  204.         serverAssertWithInfo(c,NULL,de != NULL);
  205.         clients = dictGetVal(de);
  206.         ln = listSearchKey(clients,c);
  207.         serverAssertWithInfo(c,NULL,ln != NULL);
  208.         listDelNode(clients,ln);
  209.         if (listLength(clients) == 0) {
  210.             /* Free the list and associated hash entry at all if this was
  211.              * the latest client. */
  212.             dictDelete(server.pubsub_patterns,pattern);
  213.         }
  214.     }
  215.     /* Notify the client */
  216.     if (notify) addReplyPubsubPatUnsubscribed(c,pattern);
  217.     decrRefCount(pattern);
  218.     return retval;
  219. }
  220.  
  221. /* Unsubscribe from all the channels. Return the number of channels the
  222.  * client was subscribed to. */
  223. int pubsubUnsubscribeAllChannels(client *c, int notify) {
  224.     int count = 0;
  225.     if (dictSize(c->pubsub_channels) > 0) {
  226.         dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
  227.         dictEntry *de;
  228.  
  229.         while((de = dictNext(di)) != NULL) {
  230.             robj *channel = dictGetKey(de);
  231.  
  232.             count += pubsubUnsubscribeChannel(c,channel,notify);
  233.         }
  234.         dictReleaseIterator(di);
  235.     }
  236.     /* We were subscribed to nothing? Still reply to the client. */
  237.     if (notify && count == 0) addReplyPubsubUnsubscribed(c,NULL);
  238.     return count;
  239. }
  240.  
  241. /* Unsubscribe from all the patterns. Return the number of patterns the
  242.  * client was subscribed from. */
  243. int pubsubUnsubscribeAllPatterns(client *c, int notify) {
  244.     listNode *ln;
  245.     listIter li;
  246.     int count = 0;
  247.  
  248.     listRewind(c->pubsub_patterns,&li);
  249.     while ((ln = listNext(&li)) != NULL) {
  250.         robj *pattern = ln->value;
  251.  
  252.         count += pubsubUnsubscribePattern(c,pattern,notify);
  253.     }
  254.     if (notify && count == 0) addReplyPubsubPatUnsubscribed(c,NULL);
  255.     return count;
  256. }
  257.  
  258. /* Publish a message */
  259. int pubsubPublishMessage(robj *channel, robj *message) {
  260.     int receivers = 0;
  261.     dictEntry *de;
  262.     dictIterator *di;
  263.     listNode *ln;
  264.     listIter li;
  265.  
  266.     /* Send to clients listening for that channel */
  267.     de = dictFind(server.pubsub_channels,channel);
  268.     if (de) {
  269.         list *list = dictGetVal(de);
  270.         listNode *ln;
  271.         listIter li;
  272.  
  273.         listRewind(list,&li);
  274.         while ((ln = listNext(&li)) != NULL) {
  275.             client *c = ln->value;
  276.             addReplyPubsubMessage(c,channel,message);
  277.             receivers++;
  278.         }
  279.     }
  280.     /* Send to clients listening to matching channels */
  281.     di = dictGetIterator(server.pubsub_patterns);
  282.     if (di) {
  283.         channel = getDecodedObject(channel);
  284.         while((de = dictNext(di)) != NULL) {
  285.             robj *pattern = dictGetKey(de);
  286.             list *clients = dictGetVal(de);
  287.             if (!stringmatchlen((char*)pattern->ptr,
  288.                                 sdslen(pattern->ptr),
  289.                                 (char*)channel->ptr,
  290.                                 sdslen(channel->ptr),0)) continue;
  291.  
  292.             listRewind(clients,&li);
  293.             while ((ln = listNext(&li)) != NULL) {
  294.                 client *c = listNodeValue(ln);
  295.                 addReplyPubsubPatMessage(c,pattern,channel,message);
  296.                 receivers++;
  297.             }
  298.         }
  299.         decrRefCount(channel);
  300.         dictReleaseIterator(di);
  301.     }
  302.     return receivers;
  303. }
  304.  
  305. /*-----------------------------------------------------------------------------
  306.  * Pubsub commands implementation
  307.  *----------------------------------------------------------------------------*/
  308.  
  309. /* SUBSCRIBE channel [channel ...] */
  310. void subscribeCommand(client *c) {
  311.     int j;
  312.     if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
  313.         /**
  314.          * A client that has CLIENT_DENY_BLOCKING flag on
  315.          * expect a reply per command and so can not execute subscribe.
  316.          *
  317.          * Notice that we have a special treatment for multi because of
  318.          * backword compatibility
  319.          */
  320.         addReplyError(c, "SUBSCRIBE isn't allowed for a DENY BLOCKING client");
  321.         return;
  322.     }
  323.  
  324.     for (j = 1; j < c->argc; j++)
  325.         pubsubSubscribeChannel(c,c->argv[j]);
  326.     c->flags |= CLIENT_PUBSUB;
  327. }
  328.  
  329. /* UNSUBSCRIBE [channel [channel ...]] */
  330. void unsubscribeCommand(client *c) {
  331.     if (c->argc == 1) {
  332.         pubsubUnsubscribeAllChannels(c,1);
  333.     } else {
  334.         int j;
  335.  
  336.         for (j = 1; j < c->argc; j++)
  337.             pubsubUnsubscribeChannel(c,c->argv[j],1);
  338.     }
  339.     if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
  340. }
  341.  
  342. /* PSUBSCRIBE pattern [pattern ...] */
  343. void psubscribeCommand(client *c) {
  344.     int j;
  345.     if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
  346.         /**
  347.          * A client that has CLIENT_DENY_BLOCKING flag on
  348.          * expect a reply per command and so can not execute subscribe.
  349.          *
  350.          * Notice that we have a special treatment for multi because of
  351.          * backword compatibility
  352.          */
  353.         addReplyError(c, "PSUBSCRIBE isn't allowed for a DENY BLOCKING client");
  354.         return;
  355.     }
  356.  
  357.     for (j = 1; j < c->argc; j++)
  358.         pubsubSubscribePattern(c,c->argv[j]);
  359.     c->flags |= CLIENT_PUBSUB;
  360. }
  361.  
  362. /* PUNSUBSCRIBE [pattern [pattern ...]] */
  363. void punsubscribeCommand(client *c) {
  364.     if (c->argc == 1) {
  365.         pubsubUnsubscribeAllPatterns(c,1);
  366.     } else {
  367.         int j;
  368.  
  369.         for (j = 1; j < c->argc; j++)
  370.             pubsubUnsubscribePattern(c,c->argv[j],1);
  371.     }
  372.     if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
  373. }
  374.  
  375. /* PUBLISH <channel> <message> */
  376. void publishCommand(client *c) {
  377.     int receivers = pubsubPublishMessage(c->argv[1],c->argv[2]);
  378.     if (server.cluster_enabled)
  379.         clusterPropagatePublish(c->argv[1],c->argv[2]);
  380.     else
  381.         forceCommandPropagation(c,PROPAGATE_REPL);
  382.     addReplyLongLong(c,receivers);
  383. }
  384.  
  385. /* PUBSUB command for Pub/Sub introspection. */
  386. void pubsubCommand(client *c) {
  387.     if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
  388.         const char *help[] = {
  389. "CHANNELS [<pattern>]",
  390. "    Return the currently active channels matching a <pattern> (default: '*').",
  391. "NUMPAT",
  392. "    Return number of subscriptions to patterns.",
  393. "NUMSUB [<channel> ...]",
  394. "    Return the number of subscribers for the specified channels, excluding",
  395. "    pattern subscriptions(default: no channels).",
  396. NULL
  397.         };
  398.         addReplyHelp(c, help);
  399.     } else if (!strcasecmp(c->argv[1]->ptr,"channels") &&
  400.         (c->argc == 2 || c->argc == 3))
  401.     {
  402.         /* PUBSUB CHANNELS [<pattern>] */
  403.         sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;
  404.         dictIterator *di = dictGetIterator(server.pubsub_channels);
  405.         dictEntry *de;
  406.         long mblen = 0;
  407.         void *replylen;
  408.  
  409.         replylen = addReplyDeferredLen(c);
  410.         while((de = dictNext(di)) != NULL) {
  411.             robj *cobj = dictGetKey(de);
  412.             sds channel = cobj->ptr;
  413.  
  414.             if (!pat || stringmatchlen(pat, sdslen(pat),
  415.                                        channel, sdslen(channel),0))
  416.             {
  417.                 addReplyBulk(c,cobj);
  418.                 mblen++;
  419.             }
  420.         }
  421.         dictReleaseIterator(di);
  422.         setDeferredArrayLen(c,replylen,mblen);
  423.     } else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) {
  424.         /* PUBSUB NUMSUB [Channel_1 ... Channel_N] */
  425.         int j;
  426.  
  427.         addReplyArrayLen(c,(c->argc-2)*2);
  428.         for (j = 2; j < c->argc; j++) {
  429.             list *l = dictFetchValue(server.pubsub_channels,c->argv[j]);
  430.  
  431.             addReplyBulk(c,c->argv[j]);
  432.             addReplyLongLong(c,l ? listLength(l) : 0);
  433.         }
  434.     } else if (!strcasecmp(c->argv[1]->ptr,"numpat") && c->argc == 2) {
  435.         /* PUBSUB NUMPAT */
  436.         addReplyLongLong(c,dictSize(server.pubsub_patterns));
  437.     } else {
  438.         addReplySubcommandSyntaxError(c);
  439.     }
  440. }

相关技术文章

最新源码

下载排行榜
点击QQ咨询
开通会员
返回顶部
×
微信扫码支付 元
微信扫码支付
确定支付下载
请使用微信描二维码支付
×

提示信息

×

选择支付方式

  • 微信支付
  • 支付宝付款
确定支付下载