{"subscribe",subscribeCommand,-,"rpslt",,NULL,,,,,},

    {"unsubscribe",unsubscribeCommand,-,"rpslt",,NULL,,,,,},

    {"psubscribe",psubscribeCommand,-,"rpslt",,NULL,,,,,},

    {"punsubscribe",punsubscribeCommand,-,"rpslt",,NULL,,,,,},

    {"publish",publishCommand,,"pltrF",,NULL,,,,,},

    {"pubsub",pubsubCommand,-,"pltrR",,NULL,,,,,},

// 使用 PSUBSCRIBE 订阅方式，做一层数据格式封装

typedef struct pubsubPattern {

    client *client;

    robj *pattern;

} pubsubPattern;

// 用法: SUBSCRIBE channel [channel ...]

// pubsub.c

void subscribeCommand(client *c) {

    int j;

    // n 个channel 的订阅，循环调用即可

    for (j = ; 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 = ;

    /* Add the channel to the client -> channels hash table */

    // step1. 将要订阅的 channel 添加到各自客户端的 pubsub_channels 容器中

    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {

        retval = ;

        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[]);

    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);

}

// 用法: PSUBSCRIBE pattern [pattern ...]

// pubsub.c

void psubscribeCommand(client *c) {

    int j;

    // 同样是n个channel依次注册

    for (j = ; 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 = ;

    // 直接查找对应的 pattern, 没有则添加

    if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {

        retval = ;

        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[]);

    addReply(c,shared.psubscribebulk);

    addReplyBulk(c,pattern);

    addReplyLongLong(c,clientSubscriptionsCount(c));

    return retval;

}

// 用法: PUBLISH channel message

// pubsub.c

void publishCommand(client *c) {

    // 使用 channel+message 进行发布消息

    int receivers = pubsubPublishMessage(c->argv[],c->argv[]);

    // 命令传播

    if (server.cluster_enabled)

        clusterPropagatePublish(c->argv[],c->argv[]);

    else

        forceCommandPropagation(c,PROPAGATE_REPL);

    addReplyLongLong(c,receivers);

}

// 发布一条消息

/* Publish a message */

int pubsubPublishMessage(robj *channel, robj *message) {

    int receivers = ;

    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[]);

            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),)) {

                addReply(pat->client,shared.mbulkhdr[]);

                addReply(pat->client,shared.pmessagebulk);

                addReplyBulk(pat->client,pat->pattern);

                addReplyBulk(pat->client,channel);

                addReplyBulk(pat->client,message);

                receivers++;

            }

        }

        decrRefCount(channel);

    }

    return receivers;

}

// 用法: UNSUBSCRIBE [channel [channel ...]]

// pubsub.c

void unsubscribeCommand(client *c) {

    // unsubscribe, 直接解决所有的订阅

    if (c->argc == ) {

        pubsubUnsubscribeAllChannels(c,);

    } else {

        int j;

        // 根据指定的 channel 依次解除订阅关系

        for (j = ; j < c->argc; j++)

            pubsubUnsubscribeChannel(c,c->argv[j],);

    }

    // 当一个订阅也没有，则自身不再处理 pubsub 相关的事务

    if (clientSubscriptionsCount(c) == ) 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 = ;

    /* 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 = ;

        /* 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) == ) {

            /* 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[]);

        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 = ;

    // 迭代 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 == ) {

        addReply(c,shared.mbulkhdr[]);

        addReply(c,shared.unsubscribebulk);

        addReply(c,shared.nullbulk);

        addReplyLongLong(c,dictSize(c->pubsub_channels)+

                       listLength(c->pubsub_patterns));

    }

    dictReleaseIterator(di);

    return count;

}