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中的總元素個(gè)數(shù)
    unsigned long count;        /* total count of all entries in all ziplists */
    // quicklistNodes的個(gè)數(shù)
    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 {
    // 前一個(gè)quicklistNode
    struct quicklistNode *prev; 
    // 后一個(gè)quicklistNode
    struct quicklistNode *next;
    // quicklistNode指向的ziplist
    unsigned char *zl;
    // ziplist的字節(jié)大小
    unsigned int sz;             /* ziplist size in bytes */
    // ziplist中的元素個(gè)數(shù)
    unsigned int count : 16;     /* count of items in ziplist */
    // 編碼格式，原生字節(jié)數(shù)組或壓縮存儲
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    // 存儲方式
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    // 數(shù)據(jù)是否被壓縮
    unsigned int recompress : 1; /* was this node previous compressed? */
    // 數(shù)據(jù)能否被壓縮
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    // 預(yù)留的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 {
        // 容納不了，就重新創(chuàng)建一個(gè) 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讀取數(shù)據(jù)； 
XREADGROUP：按消費(fèi)組形式讀取消息；
XPENDING和XACK：XPENDING命令可以用來查詢每個(gè)消費(fèi)組內(nèi)所有消費(fèi)者已讀取但尚未確認(rèn)的消息，而XACK命令用于向消息隊(duì)列確認(rèn)消息處理已完成。  

$ 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 {
    // 這是使用前綴樹存儲數(shù)據(jù)
    rax *rax;               /* The radix tree holding the stream. */
    uint64_t length;        /* Number of elements inside this stream. */
    // 當(dāng)前stream的最后一個(gè)id
    streamID last_id;       /* Zero if there are yet no items. */
    // 存儲當(dāng)前的消費(fèi)者組信息
    rax *cgroups;           /* Consumer groups dictionary: name -> streamCG */
} stream;

typedef struct streamID {
    // 消息創(chuàng)建時(shí)的時(shí)間
    uint64_t ms;        /* Unix time in milliseconds. */
    // 消息的序號
    uint64_t seq;       /* Sequence number. */
} streamID;

typedef struct rax {
    // radix tree 的頭節(jié)點(diǎn)
    raxNode *head;
    //  radix tree 所存儲的元素總數(shù)，每插入一個(gè) ID，計(jì)數(shù)加 1
    uint64_t numele;
    // radix tree 的節(jié)點(diǎn)總數(shù)
    uint64_t numnodes;
} rax;

typedef struct raxNode {
    // 表示從 Radix Tree 的根節(jié)點(diǎn)到當(dāng)前節(jié)點(diǎn)路徑上的字符組成的字符串，是否表示了一個(gè)完整的 key
    uint32_t iskey:1;     /* Does this node contain a key? */
    // 表明當(dāng)前key對應(yīng)的value是否為空
    uint32_t isnull:1;    /* Associated value is NULL (don't store it). */
    // 表示當(dāng)前節(jié)點(diǎn)是非壓縮節(jié)點(diǎn)，還是壓縮節(jié)點(diǎn)。
    uint32_t iscompr:1;   /* Node is compressed. */
    // 壓縮節(jié)點(diǎn)壓縮的字符串長度或者非壓縮節(jié)點(diǎn)的子節(jié)點(diǎn)個(gè)數(shù)
    uint32_t size:29;     /* Number of children, or compressed string len. */
    // 包含填充字段，同時(shí)存儲了當(dāng)前節(jié)點(diǎn)包含的字符串以及子節(jié)點(diǎn)的指針，key對應(yīng)的value指針。
    unsigned char data[];
} raxNode;

typedef struct streamCG {
    // 當(dāng)前我這個(gè)stream的最大id
    streamID last_id;
    // 還沒有收到ACK的消息列表
    rax *pel;
    // 消費(fèi)組中的所有消費(fèi)者，消費(fèi)者名稱為鍵，streamConsumer 為值 
    rax *consumers;
} streamCG;

typedef struct streamConsumer {
    // 為該消費(fèi)者最后一次活躍的時(shí)間
    mstime_t seen_time;
    // 消費(fèi)者名稱，為sds結(jié)構(gòu)
    sds name;
    // 待ACK的消息列表，和 streamCG 中指向的是同一個(gè)  
    rax *pel;
} streamConsumer;

PSUBSCRIBE pattern [pattern ...]
訂閱一個(gè)或多個(gè)符合給定模式的頻道。

PUBSUB subcommand [argument [argument ...]]
查看訂閱與發(fā)布系統(tǒng)狀態(tài)。
PUBLISH channel message
將信息發(fā)送到指定的頻道。
PUNSUBSCRIBE [pattern [pattern ...]]
退訂所有給定模式的頻道。
SUBSCRIBE channel [channel ...]
訂閱給定的一個(gè)或多個(gè)頻道的信息。
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 */
    // 保存著所有和模式相關(guān)的信息
    dict *pubsub_patterns;  /* A dict of pubsub_patterns */
    // ...
}

// 訂閱一個(gè)頻道，成功返回1，已經(jīng)訂閱返回0
int pubsubSubscribeChannel(client *c, robj *channel) {
    dictEntry *de;
    list *clients = NULL;
    int retval = 0;

    /* Add the channel to the client -> channels hash table */
    // 將頻道添加到客戶端本地的哈希表中
    // 客戶端自己也有一個(gè)訂閱頻道的列表，記錄了此客戶端所訂閱的頻道
    if (dictAdd(c->pubsub_channels,channel,NULL) == DICT_OK) {
        retval = 1;
        incrRefCount(channel);
        // 添加到服務(wù)器中的pubsub_channels中
        // 判斷下這個(gè) channel 是否已經(jīng)創(chuàng)建了
        de = dictFind(server.pubsub_channels,channel);
        if (de == NULL) {
            // 沒有創(chuàng)建，先創(chuàng)建 channel,后添加
            clients = listCreate();
            dictAdd(server.pubsub_channels,channel,clients);
            incrRefCount(channel);
        } else {
            // 已經(jīng)創(chuàng)建過了
            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所對應(yīng)的dictEntry
    de = dictFind(server.pubsub_channels,channel);
    if (de) {
        // 獲取此 channel 對應(yīng)的所有客戶端
        list *list = dictGetVal(de);
        listNode *ln;
        listIter li;

        listRewind(list,&li);
        // 一個(gè)個(gè)發(fā)送信息
        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);
             // 這里進(jìn)行匹配
            // 擁有相同的 pattern 的客戶端會被放入到同一個(gè)鏈表中
            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;
        // 客戶端端本地進(jìn)行記錄
        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) {
            // 沒有創(chuàng)建，先創(chuàng)建
            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;
}