Redis的发布订阅(Pub/Sub)功能提供了一种消息队列的实现方式,可以让发送方(发布者)向一个频道发布消息,订阅方(订阅者)则可以订阅相关频道接收消息。本文将带着大家走读源码，剖析实现机制。

redis源码版本：6.2.5

前置准备

如果不了解发布订阅(Pub/Sub)功能和具体命令，请大家先阅读 Redis发布订阅功能详解

数据结构

// path:src/server.h

// redis server 核心结构体(全局单例)
// 保存在redis服务运行的各类信息
struct redisServer {
// ...
/* Pubsub */
// 保存订阅channel 对应的客户端列表
// key:channel, val: client list
dict *pubsub_channels; /* Map channels to list of subscribed clients */

// 保存订阅pattern 对应的客户端列表
// key:pattern, val: client list
dict *pubsub_patterns; /* A dict of pubsub_patterns */
// ...
}

// 表示一个客户端，保存客户端访问信息
typedef struct client {
    // ...
    // 保存此客户端订阅了那些channel
    // key:channel, val:NULL
    dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */
    // 保存此客户端订阅了那些pattern
    list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */
    // ...
}

综上：

redisServer使用pubsub_channels保存订阅channel 对应的客户端列表
redisServer使用pubsub_patterns保存订阅pattern 对应的客户端列表
client使用pubsub_channels保存此客户端订阅了那些channel
client使用pubsub_patterns保存此客户端订阅了那些pattern

订阅&取消订阅

// path:src/pubsub.c
/*-----------------------------------------------------------------------------
 * Pubsub low level API
 *----------------------------------------------------------------------------*/

/* Return the number of channels + patterns a client is subscribed to. */
int clientSubscriptionsCount(client *c) {
    return dictSize(c->pubsub_channels)+
           listLength(c->pubsub_patterns);
}

/* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
 * 0 if the client was already subscribed to that channel. */
// 订阅channel
int pubsubSubscribeChannel(client *c, robj *channel) {
    dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    /* Add the channel to the client -> channels hash table */
    // 尝试将channel 加到client 的pubsub_channels中
    // 加入失败，说明原本client 已经订阅的channel
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        incrRefCount(channel);
        /* Add the client to the channel -> list of clients hash table */
        // 找到server.pubsub_channels 对应的channel 列表
        de = dictFind(server.pubsub_channels,channel);
        if (de == NULL) {
            clients = listCreate();
            dictAdd(server.pubsub_channels,channel,clients);
            incrRefCount(channel);
        } else {
            clients = dictGetVal(de);
        }
        // 将client 加到server.pubsub_channels 对应的channel 列表中
        listAddNodeTail(clients,c);
    }
    /* Notify the client */
    addReplyPubsubSubscribed(c,channel);
    return retval;
}

/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
 * 0 if the client was not subscribed to the specified channel. */
// 取消订阅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... */
    // 尝试从client 的pubsub_channels中 删除channel
    // 删除失败，说明client 没有订阅channel
    if (dictDelete(c->pubsub_channels,channel) == DICT_OK) {
        retval = 1;
        /* Remove the client from the channel -> clients list hash table */
        // 找到server.pubsub_channels 对应channel 的client列表
        de = dictFind(server.pubsub_channels,channel);
        serverAssertWithInfo(c,NULL,de != NULL);

        // 从列表中删除client
        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. */
            // channel 对应列表已经没有订阅的客户端了, 清理对应key&val
            dictDelete(server.pubsub_channels,channel);
        }
    }
    /* Notify the client */
    // 通知被取消订阅的客户端
    if (notify) addReplyPubsubUnsubscribed(c,channel);
    decrRefCount(channel); /* it is finally safe to release it */
    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. */
// 订阅Pattern
int pubsubSubscribePattern(client *c, robj *pattern) {
    dictEntry *de;
    list *clients;
    int retval = 0;

    // 尝试在c->pubsub_patterns查找对应pattern
    // 查到到了，说明已经订阅了pattern，无需再处理
    if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
        retval = 1;
        // 还没订阅, 将pattern 加到c->pubsub_patterns中
        listAddNodeTail(c->pubsub_patterns,pattern);
        incrRefCount(pattern);
        /* Add the client to the pattern -> list of clients hash table */
        // 查找 server.pubsub_patterns 对应pattern的客户端列表
        de = dictFind(server.pubsub_patterns,pattern);
        if (de == NULL) {
            clients = listCreate();
            dictAdd(server.pubsub_patterns,pattern,clients);
            incrRefCount(pattern);
        } else {
            clients = dictGetVal(de);
        }
        // 将client加到server.pubsub_patterns 对应pattern的客户端列表中
        listAddNodeTail(clients,c);
    }
    /* Notify the client */
    // 通知客户端
    addReplyPubsubPatSubscribed(c,pattern);
    return retval;
}

/* Unsubscribe a client from a channel. Returns 1 if the operation succeeded, or
 * 0 if the client was not subscribed to the specified channel. */
// 取消订阅Pattern
int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
    dictEntry *de;
    list *clients;
    listNode *ln;
    int retval = 0;

    incrRefCount(pattern); /* Protect the object. May be the same we remove */
    // 尝试在c->pubsub_patterns查找对应pattern
    // 查不到，说明已经取消订阅或者没有订阅pattern，无需再处理
    if ((ln = listSearchKey(c->pubsub_patterns,pattern)) != NULL) {
        retval = 1;
        // 将pattern 从c->pubsub_patterns中删除
        listDelNode(c->pubsub_patterns,ln);
        /* Remove the client from the pattern -> clients list hash table */
         // 查找 server.pubsub_patterns 对应pattern的客户端列表
        de = dictFind(server.pubsub_patterns,pattern);
        serverAssertWithInfo(c,NULL,de != NULL);
        clients = dictGetVal(de);
        ln = listSearchKey(clients,c);
        serverAssertWithInfo(c,NULL,ln != NULL);
        // 从对应pattern的客户端列表上次client
        listDelNode(clients,ln);
        if (listLength(clients) == 0) {
            /* Free the list and associated hash entry at all if this was
             * the latest client. */
            dictDelete(server.pubsub_patterns,pattern);
        }
    }
    /* Notify the client */
    // 通知client
    if (notify) addReplyPubsubPatUnsubscribed(c,pattern);
    decrRefCount(pattern);
    return retval;
}

/* Unsubscribe from all the channels. Return the number of channels the
 * client was subscribed to. */
// 取消所有订阅的channels
int pubsubUnsubscribeAllChannels(client *c, int notify) {
    int count = 0;
    if (dictSize(c->pubsub_channels) > 0) {
        dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
        dictEntry *de;
        // 遍历c->pubsub_channels
        while((de = dictNext(di)) != NULL) {
            robj *channel = dictGetKey(de);
            // 调用pubsubUnsubscribeChannel， 逐个取消订阅
            count += pubsubUnsubscribeChannel(c,channel,notify);
        }
        dictReleaseIterator(di);
    }
    /* We were subscribed to nothing? Still reply to the client. */
    if (notify && count == 0) addReplyPubsubUnsubscribed(c,NULL);
    return count;
}

/* Unsubscribe from all the patterns. Return the number of patterns the
 * client was subscribed from. */
// 取消所有订阅的patterns
int pubsubUnsubscribeAllPatterns(client *c, int notify) {
    listNode *ln;
    listIter li;
    int count = 0;

    listRewind(c->pubsub_patterns,&li);
    // 遍历c->pubsub_patterns
    while ((ln = listNext(&li)) != NULL) {
        robj *pattern = ln->value;
        // 调用pubsubUnsubscribePattern 逐个取消pattern订阅
        count += pubsubUnsubscribePattern(c,pattern,notify);
    }
    if (notify && count == 0) addReplyPubsubPatUnsubscribed(c,NULL);
    return count;
}


/*-----------------------------------------------------------------------------
 * Pubsub commands implementation
 *----------------------------------------------------------------------------*/

/* SUBSCRIBE channel [channel ...] */
void subscribeCommand(client *c) {
    int j;
    if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
        /**
         * A client that has CLIENT_DENY_BLOCKING flag on
         * expect a reply per command and so can not execute subscribe.
         *
         * Notice that we have a special treatment for multi because of
         * backword compatibility
         */
        // 处于事务或者block中，不能订阅
        addReplyError(c, "SUBSCRIBE isn't allowed for a DENY BLOCKING client");
        return;
    }

    // 遍历参数中的channel, 分别订阅
    for (j = 1; j < c->argc; j++)
        pubsubSubscribeChannel(c,c->argv[j]);
    c->flags |= CLIENT_PUBSUB;
}

/* UNSUBSCRIBE [channel [channel ...]] */
void unsubscribeCommand(client *c) {
    if (c->argc == 1) {
        // 没有多个参数，取消所有订阅
        pubsubUnsubscribeAllChannels(c,1);
    } else {
        int j;
        // 带了channel,逐个取消订阅
        for (j = 1; j < c->argc; j++)
            pubsubUnsubscribeChannel(c,c->argv[j],1);
    }
    if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
}

/* PSUBSCRIBE pattern [pattern ...] */
void psubscribeCommand(client *c) {
    int j;
    if ((c->flags & CLIENT_DENY_BLOCKING) && !(c->flags & CLIENT_MULTI)) {
        /**
         * A client that has CLIENT_DENY_BLOCKING flag on
         * expect a reply per command and so can not execute subscribe.
         *
         * Notice that we have a special treatment for multi because of
         * backword compatibility
         */
        // 处于事务或者block中，不能订阅
        addReplyError(c, "PSUBSCRIBE isn't allowed for a DENY BLOCKING client");
        return;
    }

    // 逐个订阅pattern
    for (j = 1; j < c->argc; j++)
        pubsubSubscribePattern(c,c->argv[j]);
    c->flags |= CLIENT_PUBSUB;
}

/* PUNSUBSCRIBE [pattern [pattern ...]] */
void punsubscribeCommand(client *c) {
    if (c->argc == 1) {
        // 没带pattern 参数，取消所有pattern 订阅
        pubsubUnsubscribeAllPatterns(c,1);
    } else {
        int j;
        // 带了pattern, 逐个取消订阅
        for (j = 1; j < c->argc; j++)
            pubsubUnsubscribePattern(c,c->argv[j],1);
    }
    if (clientSubscriptionsCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
}

订阅(channel或者pattern)时，都是把对应channel或者pattern保存到client和redisServer结构体中
取消订阅时，都是从client和redisServer结构体移除相关数据

发布消息(publish)

// path:src/pubsub.c
/* Publish a message */
// 具体处理消息发布
int pubsubPublishMessage(robj *channel, robj *message) {
    int receivers = 0;
    dictEntry *de;
    dictIterator *di;
    listNode *ln;
    listIter li;

    /* Send to clients listening for that channel */
    // 从server.pubsub_channels 获取channel对应的client列表
    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;
            // 向每个客户端发送消息
            addReplyPubsubMessage(c,channel,message);
            receivers++;
        }
    }
    /* Send to clients listening to matching channels */
    di = dictGetIterator(server.pubsub_patterns);
    if (di) {
        channel = getDecodedObject(channel);
        // 遍历server.pubsub_patterns 列表
        while((de = dictNext(di)) != NULL) {
            robj *pattern = dictGetKey(de);
            list *clients = dictGetVal(de);
            // 判断pattern 是否与发布消息的channel 匹配
            if (!stringmatchlen((char*)pattern->ptr,
                                sdslen(pattern->ptr),
                                (char*)channel->ptr,
                                sdslen(channel->ptr),0)) continue;

            listRewind(clients,&li);
            // 遍历匹配pattern 的订阅客户端列表
            while ((ln = listNext(&li)) != NULL) {
                client *c = listNodeValue(ln);
                // 向每个客户端发送消息
                addReplyPubsubPatMessage(c,pattern,channel,message);
                receivers++;
            }
        }
        decrRefCount(channel);
        dictReleaseIterator(di);
    }
    return receivers;
}

/* PUBLISH <channel> <message> */
// 向某个chanel 发布消息
void publishCommand(client *c) {
    // 调用pubsubPublishMessage 具体处理消息发布
    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);
}