Redis 消息队列

127.0.0.1:6379> LPUSH test "ceshi-1"
(integer) 1

127.0.0.1:6379> RPOP test
"ceshi-1"

127.0.0.1:6379> BRPOP test 10

127.0.0.1:6379> LPUSH test "ceshi-1"
(integer) 1
127.0.0.1:6379> LPUSH test "ceshi-2"
(integer) 2
127.0.0.1:6379> BRPOPLPUSH test a-test 100
"ceshi-1"
127.0.0.1:6379> BRPOPLPUSH test a-test 100
"ceshi-2"
127.0.0.1:6379> BRPOPLPUSH test a-test 100

127.0.0.1:6379> RPOP a-test
"ceshi-1"
127.0.0.1:6379> RPOP a-test
"ceshi-2"

void listTypePush(robj *subject, robj *value, int where) {
    if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
        int pos = (where == LIST_HEAD) ? QUICKLIST_HEAD : QUICKLIST_TAIL;
        if (value->encoding == OBJ_ENCODING_INT) {
            char buf[32];
            ll2string(buf, 32, (long)value->ptr);
            quicklistPush(subject->ptr, buf, strlen(buf), pos);
        } else {
            quicklistPush(subject->ptr, value->ptr, sdslen(value->ptr), pos);
        }
    } else {
        serverPanic("Unknown list encoding");
    }
}

/* Wrapper to allow argument-based switching between HEAD/TAIL pop */
void quicklistPush(quicklist *quicklist, void *value, const size_t sz,
                   int where) {
    if (where == QUICKLIST_HEAD) {
        quicklistPushHead(quicklist, value, sz);
    } else if (where == QUICKLIST_TAIL) {
        quicklistPushTail(quicklist, value, sz);
    }
}

typedef struct quicklist {
    // quicklist的链表头
    quicklistNode *head;
    // quicklist的链表尾
    quicklistNode *tail;
    // 所有ziplist中的总元素个数
    unsigned long count;        /* total count of all entries in all ziplists */
    // quicklistNodes的个数
    unsigned long len;          /* number of quicklistNodes */
    int fill : QL_FILL_BITS;              /* fill factor for individual nodes */
    unsigned int compress : QL_COMP_BITS; /* depth of end nodes not to compress;0=off */
    unsigned int bookmark_count: QL_BM_BITS;
    quicklistBookmark bookmarks[];
} quicklist;

typedef struct quicklistNode {
    // 前一个quicklistNode
    struct quicklistNode *prev; 
    // 后一个quicklistNode
    struct quicklistNode *next;
    // quicklistNode指向的ziplist
    unsigned char *zl;
    // ziplist的字节大小
    unsigned int sz;             /* ziplist size in bytes */
    // ziplist中的元素个数
    unsigned int count : 16;     /* count of items in ziplist */
    // 编码格式，原生字节数组或压缩存储
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    // 存储方式
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    // 数据是否被压缩
    unsigned int recompress : 1; /* was this node previous compressed? */
    // 数据能否被压缩
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    // 预留的bit位
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

/* Add new entry to head node of quicklist.
 *
 * Returns 0 if used existing head.
 * Returns 1 if new head created. */
int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_head = quicklist->head;
    assert(sz < UINT32_MAX); /* TODO: add support for quicklist nodes that are sds encoded (not zipped) */
    if (likely(
            // 检测插入位置的 ziplist 是否能容纳该元素
            _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        quicklist->head->zl =
            ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
        quicklistNodeUpdateSz(quicklist->head);
    } else {
        // 容纳不了，就重新创建一个 quicklistNode
        quicklistNode *node = quicklistCreateNode();
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);

        quicklistNodeUpdateSz(node);
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
    }
    quicklist->count++;
    quicklist->head->count++;
    return (orig_head != quicklist->head);
}

XADD：插入消息，保证有序，可以自动生成全局唯一ID；

XREAD：用于读取消息，可以按ID读取数据； 
XREADGROUP：按消费组形式读取消息；
XPENDING和XACK：XPENDING命令可以用来查询每个消费组内所有消费者已读取但尚未确认的消息，而XACK命令用于向消息队列确认消息处理已完成。  

$ XADD key ID field value [field value ...]

$ XADD teststream * name xiaohong surname xiaobai
"1646650328883-0"

$ XREAD [COUNT count] [BLOCK milliseconds] STREAMS key [key ...] id [id ...]

$ XREAD BLOCK 100 STREAMS  teststream 0
1) 1) "teststream"
   2) 1) 1) "1646650328883-0"
         2) 1) "name"
            2) "xiaohong"
            3) "surname"
            4) "xiaobai"

$ XGROUP [CREATE key groupname id-or-$] [SETID key groupname id-or-$] [DESTROY key groupname] [DELCONSUMER key groupname consumername]

$ XGROUP CREATE teststream test-consumer-group-name 0-0  

$ XGROUP CREATE teststream test-consumer-group-name $

$ XREADGROUP GROUP group consumer [COUNT count] [BLOCK milliseconds] [NOACK] STREAMS key [key ...] ID [ID ...]

$ XADD teststream * name xiaohong surname xiaobai
"1646653392799-0"

$ XREADGROUP GROUP test-consumer-group-name test-consumer-name COUNT 1 STREAMS teststream >
1) 1) "teststream"
   2) 1) 1) "1646653392799-0"
         2) 1) "name"
            2) "xiaohong"
            3) "surname"
            4) "xiaobai"

$ XPENDING teststream test-consumer-group-name
1) (integer) 3
2) "1646653325535-0"
3) "1646653392799-0"
4) 1) 1) "test-consumer-name"
      2) "3"

typedef struct stream {
    // 这是使用前缀树存储数据
    rax *rax;               /* The radix tree holding the stream. */
    uint64_t length;        /* Number of elements inside this stream. */
    // 当前stream的最后一个id
    streamID last_id;       /* Zero if there are yet no items. */
    // 存储当前的消费者组信息
    rax *cgroups;           /* Consumer groups dictionary: name -> streamCG */
} stream;

typedef struct streamID {
    // 消息创建时的时间
    uint64_t ms;        /* Unix time in milliseconds. */
    // 消息的序号
    uint64_t seq;       /* Sequence number. */
} streamID;

typedef struct rax {
    // radix tree 的头节点
    raxNode *head;
    //  radix tree 所存储的元素总数，每插入一个 ID，计数加 1
    uint64_t numele;
    // radix tree 的节点总数
    uint64_t numnodes;
} rax;

typedef struct raxNode {
    // 表示从 Radix Tree 的根节点到当前节点路径上的字符组成的字符串，是否表示了一个完整的 key
    uint32_t iskey:1;     /* Does this node contain a key? */
    // 表明当前key对应的value是否为空
    uint32_t isnull:1;    /* Associated value is NULL (don't store it). */
    // 表示当前节点是非压缩节点，还是压缩节点。
    uint32_t iscompr:1;   /* Node is compressed. */
    // 压缩节点压缩的字符串长度或者非压缩节点的子节点个数
    uint32_t size:29;     /* Number of children, or compressed string len. */
    // 包含填充字段，同时存储了当前节点包含的字符串以及子节点的指针，key对应的value指针。
    unsigned char data[];
} raxNode;

typedef struct streamCG {
    // 当前我这个stream的最大id
    streamID last_id;
    // 还没有收到ACK的消息列表
    rax *pel;
    // 消费组中的所有消费者，消费者名称为键，streamConsumer 为值 
    rax *consumers;
} streamCG;

typedef struct streamConsumer {
    // 为该消费者最后一次活跃的时间
    mstime_t seen_time;
    // 消费者名称，为sds结构
    sds name;
    // 待ACK的消息列表，和 streamCG 中指向的是同一个  
    rax *pel;
} streamConsumer;

PSUBSCRIBE pattern [pattern ...]
订阅一个或多个符合给定模式的频道。

PUBSUB subcommand [argument [argument ...]]
查看订阅与发布系统状态。
PUBLISH channel message
将信息发送到指定的频道。
PUNSUBSCRIBE [pattern [pattern ...]]
退订所有给定模式的频道。
SUBSCRIBE channel [channel ...]
订阅给定的一个或多个频道的信息。
UNSUBSCRIBE [channel [channel ...]]
指退订给定的频道。

$ SUBSCRIBE test

$ PUBLISH test 1

$ psubscribe p-test*

$ PUBLISH p-testa ceshi-1

struct redisServer {
    // 保存订阅频道的信息
    dict *pubsub_channels;  /* Map channels to list of subscribed clients */
    // 保存着所有和模式相关的信息
    dict *pubsub_patterns;  /* A dict of pubsub_patterns */
    // ...
}

// 订阅一个频道，成功返回1，已经订阅返回0
int pubsubSubscribeChannel(client *c, robj *channel) {
    dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    /* Add the channel to the client -> channels hash table */
    // 将频道添加到客户端本地的哈希表中
    // 客户端自己也有一个订阅频道的列表，记录了此客户端所订阅的频道
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        incrRefCount(channel);
        // 添加到服务器中的pubsub_channels中
        // 判断下这个 channel 是否已经创建了
        de = dictFind(server.pubsub_channels,channel);
        if (de == NULL) {
            // 没有创建，先创建 channel,后添加
            clients = listCreate();
            dictAdd(server.pubsub_channels,channel,clients);
            incrRefCount(channel);
        } else {
            // 已经创建过了
            clients = dictGetVal(de);
        }
        // 在尾部添加客户端
        listAddNodeTail(clients,c);
    }
    /* Notify the client */
    addReplyPubsubSubscribed(c,channel);
}

typedef struct client {
    dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
    list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
} client;

// 取消 client 订阅
int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
    dictEntry *de;
    list *clients;
    listNode *ln;
    int retval = 0;

    // 客户端在本地的哈希表中删除channel
    incrRefCount(channel); /* channel may be just a pointer to the same object
                            we have in the hash tables. Protect it... */
    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 */
    if (notify) addReplyPubsubUnsubscribed(c,channel);
    decrRefCount(channel); /* it is finally safe to release it */
    return retval;
}

/* 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 */
    // 找到Channel所对应的dictEntry
    de = dictFind(server.pubsub_channels,channel);
    if (de) {
        // 获取此 channel 对应的所有客户端
        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);
        while((de = dictNext(di)) != NULL) {
            robj *pattern = dictGetKey(de);
            list *clients = dictGetVal(de);
             // 这里进行匹配
            // 拥有相同的 pattern 的客户端会被放入到同一个链表中
            if (!stringmatchlen((char*)pattern->ptr,
                                sdslen(pattern->ptr),
                                (char*)channel->ptr,
                                sdslen(channel->ptr),0)) continue;

            listRewind(clients,&li);
            while ((ln = listNext(&li)) != NULL) {
                client *c = listNodeValue(ln);
                addReplyPubsubPatMessage(c,pattern,channel,message);
                receivers++;
            }
        }
        decrRefCount(channel);
        dictReleaseIterator(di);
    }
    return receivers;
}

/* 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) {
    dictEntry *de;
    list *clients;
    int retval = 0;

    // 如果客户端没有订阅过
    if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
        retval = 1;
        // 客户端端本地进行记录
        listAddNodeTail(c->pubsub_patterns,pattern);
        incrRefCount(pattern);
        /* Add the client to the pattern -> list of clients hash table */
        de = dictFind(server.pubsub_patterns,pattern);
        if (de == NULL) {
            // 没有创建，先创建
            clients = listCreate();
            dictAdd(server.pubsub_patterns,pattern,clients);
            incrRefCount(pattern);
        } else {
            clients = dictGetVal(de);
        }
        listAddNodeTail(clients,c);
    }
    /* Notify the client */
    addReplyPubsubPatSubscribed(c,pattern);
    return retval;
}