Redis原始碼剖析—連結串列結構

1. redis中的連結串列

在redis中連結串列的應用非常廣泛，例如列表鍵的底層實現之一就是連結串列。而且，在redis中的連結串列結構被實現成為雙向連結串列，因此，在頭部和尾部進行的操作就會非常快。

通過列表鍵的命令感受一下雙向連結串列：列表鍵命令詳解

127.0.0.1:6379> LPUSH list a b c    //依次在連結串列頭部插入a、b、c
(integer) 3
127.0.0.1:6379> RPUSH list d e f    //依次在連結串列尾部插入d、e、f
(integer) 6
127.0.0.1:6379
 
> LRANGE list 0 -1    //檢視list的值
1) "c"
2) "b"
3) "a"
4) "d"
5) "e"
6) "f"

2. 連結串列的實現

2.1 連結串列節點的實現

每個連結串列節點由adlist.h/listNode來表示

typedef struct listNode {
    struct listNode *prev; //前驅節點，如果是list的頭結點，則prev指向NULL
    struct listNode *next;//後繼節點，如果是list尾部結點，則next指向NULL
    void *value;            //萬能指標，能夠存放任何資訊
 

} listNode;

listNode結構通過prev和next指標就組成了雙向連結串列。剛才通過列表鍵生成的雙向連結串列如下圖：

使用雙向連結串列的好處：

• prev和next指標：獲取某個節點的前驅節點和後繼節點複雜度為O(1)。

2.2 表頭的實現

redis還提供了一個表頭，用於存放上面雙向連結串列的資訊，它由adlist.h/list結構表示：

typedef struct list {
    listNode *head;     //連結串列頭結點指標
    listNode *tail;     //連結串列尾結點指標

    //下面的三個函式指標就像類中的成員函式一樣
 

    void *(*dup)(void *ptr);    //複製連結串列節點儲存的值
    void (*free)(void *ptr);    //釋放連結串列節點儲存的值
    int (*match)(void *ptr, void *key); //比較連結串列節點所儲存的節點值和另一個輸入的值是否相等
    unsigned long len;      //連結串列長度計數器
} list;

將表頭和雙向連結串列連線起來，如圖：

利用list表頭管理連結串列資訊的好處：

• head和tail指標：對於連結串列的頭結點和尾結點操作的複雜度為O(1)。
• len 連結串列長度計數器：獲取連結串列中節點數量的複雜度為O(1)。
• dup、free和match指標：實現多型，連結串列節點listNode使用萬能指標void *儲存節點的值，而表頭list使用dup、free和match指標來針對連結串列中存放的不同物件從而實現不同的方法。

3. 連結串列結構原始碼剖析

3.1 adlist.h檔案

針對list結構和listNode結構的賦值和查詢操作使用巨集進行封裝，而且一下操作的複雜度均為O(1)。

#define listLength(l) ((l)->len)                    //返回連結串列l節點數量
#define listFirst(l) ((l)->head)                    //返回連結串列l的頭結點地址
#define listLast(l) ((l)->tail)                     //返回連結串列l的尾結點地址
#define listPrevNode(n) ((n)->prev)                 //返回節點n的前驅節點地址
#define listNextNode(n) ((n)->next)                 //返回節點n的後繼節點地址
#define listNodeValue(n) ((n)->value)               //返回節點n的節點值

#define listSetDupMethod(l,m) ((l)->dup = (m))      //設定連結串列l的複製函式為m方法
#define listSetFreeMethod(l,m) ((l)->free = (m))    //設定連結串列l的釋放函式為m方法
#define listSetMatchMethod(l,m) ((l)->match = (m))  //設定連結串列l的比較函式為m方法

#define listGetDupMethod(l) ((l)->dup)              //返回連結串列l的賦值函式
#define listGetFree(l) ((l)->free)                  //返回連結串列l的釋放函式
#define listGetMatchMethod(l) ((l)->match)          //返回連結串列l的比較函式

連結串列操作的函式原型（Prototypes）：

list *listCreate(void);                                 //建立一個表頭
void listRelease(list *list);                           //釋放list表頭和連結串列
list *listAddNodeHead(list *list, void *value);         //將value新增到list連結串列的頭部
list *listAddNodeTail(list *list, void *value);         //將value新增到list連結串列的尾部
list *listInsertNode(list *list, listNode *old_node, void *value, int after);//在list中，根據after在old_node節點前後插入值為value的節點。
void listDelNode(list *list, listNode *node);           //從list刪除node節點

listIter *listGetIterator(list *list, int direction);   //為list建立一個迭代器iterator
listNode *listNext(listIter *iter);                     //返回迭代器iter指向的當前節點並更新iter       
void listReleaseIterator(listIter *iter);               //釋放iter迭代器

list *listDup(list *orig);                              //拷貝表頭為orig的連結串列並返回
listNode *listSearchKey(list *list, void *key);         //在list中查詢value為key的節點並返回
listNode *listIndex(list *list, long index);            //返回下標為index的節點地址
void listRewind(list *list, listIter *li);              //將迭代器li重置為list的頭結點並且設定為正向迭代
void listRewindTail(list *list, listIter *li);          //將迭代器li重置為list的尾結點並且設定為反向迭代
void listRotate(list *list);                            //將尾節點插到頭結點

3.2 連結串列迭代器

在adlist.h檔案中，使用C語言實現了迭代器，原始碼如下：

typedef struct listIter {
    listNode *next;     //迭代器當前指向的節點（名字叫next有點迷惑）
    int direction;      //迭代方向，可以取以下兩個值：AL_START_HEAD和AL_START_TAIL
} listIter

#define AL_START_HEAD 0 //正向迭代：從表頭向表尾進行迭代
#define AL_START_TAIL 1 //反向迭代：從表尾到表頭進行迭代

在listDup函式中就使用了迭代器，listDup函式的定義如下：

//listDup的功能是拷貝一份連結串列
list *listDup(list *orig)
{
    list *copy;
    listIter *iter;
    listNode *node;

    if ((copy = listCreate()) == NULL)  //建立一個表頭
        return NULL;

    //設定新建表頭的處理函式
    copy->dup = orig->dup;
    copy->free = orig->free;
    copy->match = orig->match;

    //迭代整個orig的連結串列，重點關注此部分。
    iter = listGetIterator(orig, AL_START_HEAD);//為orig定義一個迭代器並設定迭代方向，在c++中例如是  vector<int>::interator it;
    while((node = listNext(iter)) != NULL) {    //迭代器根據迭代方向不停迭代，相當於++it
        void *value;

        //複製節點值到新節點
        if (copy->dup) {    //如果定義了list結構中的dup指標，則使用該方法拷貝節點值。
            value = copy->dup(node->value);
            if (value == NULL) {
                listRelease(copy);
                listReleaseIterator(iter);
                return NULL;
            }
        } else
            value = node->value;    //獲得當前node的value值

        if (listAddNodeTail(copy, value) == NULL) { //將node節點尾插到copy表頭的連結串列中
            listRelease(copy);
            listReleaseIterator(iter);
            return NULL;
        }
    }

    listReleaseIterator(iter);    //自行釋放迭代器
    return copy;    //返回拷貝副本
}

迭代器的好處：

• 提供一種方法順序訪問一個聚合物件中各個元素, 而又不需暴露該物件的內部表示。
• 將指標操作進行了統一封裝，程式碼可讀性增強。

3.3 adlist.c檔案

剛才所有函式的定義如下：

list *listCreate(void)  //建立一個表頭
{
    struct list *list;

    //為表頭分配記憶體
    if ((list = zmalloc(sizeof(*list))) == NULL)
        return NULL;
    //初始化表頭
    list->head = list->tail = NULL;
    list->len = 0;
    list->dup = NULL;
    list->free = NULL;
    list->match = NULL;

    return list;    //返回表頭
}

/* Free the whole list.
 *
 * This function can't fail. */
void listRelease(list *list)    //釋放list表頭和連結串列
{
    unsigned long len;
    listNode *current, *next;

    current = list->head;   //備份頭節點地址
    len = list->len;        //備份連結串列元素個數，使用備份操作防止更改原有資訊
    while(len--) {          //遍歷連結串列
        next = current->next;
        if (list->free) list->free(current->value); //如果設定了list結構的釋放函式，則呼叫該函式釋放節點值
        zfree(current);
        current = next;
    }
    zfree(list);    //最後釋放表頭
}

/* Add a new node to the list, to head, containing the specified 'value'
 * pointer as value.
 *
 * On error, NULL is returned and no operation is performed (i.e. the
 * list remains unaltered).
 * On success the 'list' pointer you pass to the function is returned. */
list *listAddNodeHead(list *list, void *value)  //將value新增到list連結串列的頭部
{
    listNode *node;

    if ((node = zmalloc(sizeof(*node))) == NULL)    //為新節點分配空間
        return NULL;
    node->value = value;    //設定node的value值

    if (list->len == 0) {   //將node頭插到空連結串列
        list->head = list->tail = node;
        node->prev = node->next = NULL;
    } else {                //將node頭插到非空連結串列
        node->prev = NULL;
        node->next = list->head;
        list->head->prev = node;
        list->head = node;
    }

    list->len++;    //連結串列元素計數器加1

    return list;
}

/* Add a new node to the list, to tail, containing the specified 'value'
 * pointer as value.
 *
 * On error, NULL is returned and no operation is performed (i.e. the
 * list remains unaltered).
 * On success the 'list' pointer you pass to the function is returned. */
list *listAddNodeTail(list *list, void *value)  //將value新增到list連結串列的尾部
{
    listNode *node;

    if ((node = zmalloc(sizeof(*node))) == NULL)    //為新節點分配空間
        return NULL;
    node->value = value;    //設定node的value值
    if (list->len == 0) {   //將node尾插到空連結串列
        list->head = list->tail = node;
        node->prev = node->next = NULL;
    } else {                //將node頭插到非空連結串列
        node->prev = list->tail;
        node->next = NULL;
        list->tail->next = node;
        list->tail = node;
    }
    list->len++;    //更新連結串列節點計數器

    return list;
}

list *listInsertNode(list *list, listNode *old_node, void *value, int after)    //在list中，根據after在old_node節點前後插入值為value的節點。
{
    listNode *node;

    if ((node = zmalloc(sizeof(*node))) == NULL) //為新節點分配空間
        return NULL;
    node->value = value;    //設定node的value值

    if (after) {    //after 非零，則將節點插入到old_node的後面
        node->prev = old_node;
        node->next = old_node->next;
        if (list->tail == old_node) {   //目標節點如果是連結串列的尾節點，更新list的tail指標
            list->tail = node;
        }
    } else {        //after 為零，則將節點插入到old_node的前面
        node->next = old_node;
        node->prev = old_node->prev;
        if (list->head == old_node) {   //如果節點如果是連結串列的頭節點，更新list的head指標
            list->head = node;
        }
    }
    if (node->prev != NULL) {   //如果有，則更新node的前驅節點的指標
        node->prev->next = node;
    }
    if (node->next != NULL) {   //如果有，則更新node的後繼節點的指標
        node->next->prev = node;
    }
    list->len++;    //更新連結串列節點計數器
    return list;
}

/* Remove the specified node from the specified list.
 * It's up to the caller to free the private value of the node.
 *
 * This function can't fail. */
void listDelNode(list *list, listNode *node)    //從list刪除node節點
{
    if (node->prev) //更新node的前驅節點的指標
        node->prev->next = node->next;
    else
        list->head = node->next;
    if (node->next) //更新node的後繼節點的指標
        node->next->prev = node->prev;
    else
        list->tail = node->prev;

    if (list->free) list->free(node->value);    //如果設定了list結構的釋放函式，則呼叫該函式釋放節點值
    zfree(node);    //釋放節點
    list->len--;    //更新連結串列節點計數器
}

/* Returns a list iterator 'iter'. After the initialization every
 * call to listNext() will return the next element of the list.
 *
 * This function can't fail. */
listIter *listGetIterator(list *list, int direction)    //為list建立一個迭代器iterator
{
    listIter *iter;

    if ((iter = zmalloc(sizeof(*iter))) == NULL) return NULL;   //為迭代器申請空間
    if (direction == AL_START_HEAD)     //設定迭代指標的起始位置
        iter->next = list->head;
    else
        iter->next = list->tail;
    iter->direction = direction;        //設定迭代方向
    return iter;
}

/* Release the iterator memory */
void listReleaseIterator(listIter *iter) {  //釋放iter迭代器
    zfree(iter);
}

/* Create an iterator in the list private iterator structure */
void listRewind(list *list, listIter *li) { //將迭代器li重置為list的頭結點並且設定為正向迭代
    li->next = list->head;              //設定迭代指標的起始位置
    li->direction = AL_START_HEAD;      //設定迭代方向從頭到尾
}

void listRewindTail(list *list, listIter *li) { //將迭代器li重置為list的尾結點並且設定為反向迭代
    li->next = list->tail;              //設定迭代指標的起始位置
    li->direction = AL_START_TAIL;      //設定迭代方向從尾到頭
}

/* Return the next element of an iterator.
 * It's valid to remove the currently returned element using
 * listDelNode(), but not to remove other elements.
 *
 * The function returns a pointer to the next element of the list,
 * or NULL if there are no more elements, so the classical usage patter
 * is:
 *
 * iter = listGetIterator(list,<direction>);
 * while ((node = listNext(iter)) != NULL) {
 *     doSomethingWith(listNodeValue(node));
 * }
 *
 * */
listNode *listNext(listIter *iter)  //返回迭代器iter指向的當前節點並更新iter
{
    listNode *current = iter->next; //備份當前迭代器指向的節點

    if (current != NULL) {
        if (iter->direction == AL_START_HEAD)   //根據迭代方向更新迭代指標
            iter->next = current->next;
        else
            iter->next = current->prev;
    }
    return current;     //返回備份的當前節點地址
}

/* Duplicate the whole list. On out of memory NULL is returned.
 * On success a copy of the original list is returned.
 *
 * The 'Dup' method set with listSetDupMethod() function is used
 * to copy the node value. Otherwise the same pointer value of
 * the original node is used as value of the copied node.
 *
 * The original list both on success or error is never modified. */
list *listDup(list *orig)   //拷貝表頭為orig的連結串列並返回
{
    list *copy;
    listIter *iter;
    listNode *node;

    if ((copy = listCreate()) == NULL)  //建立一個表頭
        return NULL;

    //設定新建表頭的處理函式
    copy->dup = orig->dup;
    copy->free = orig->free;
    copy->match = orig->match;
    //迭代整個orig的連結串列
    iter = listGetIterator(orig, AL_START_HEAD);    //為orig定義一個迭代器並設定迭代方向
    while((node = listNext(iter)) != NULL) {    //迭代器根據迭代方向不停迭代
        void *value;

        //複製節點值到新節點
        if (copy->dup) {
            value = copy->dup(node->value); //如果定義了list結構中的dup指標，則使用該方法拷貝節點值。
            if (value == NULL) {
                listRelease(copy);
                listReleaseIterator(iter);
                return NULL;
            }
        } else
            value = node->value;    //獲得當前node的value值

        if (listAddNodeTail(copy, value) == NULL) { //將node節點尾插到copy表頭的連結串列中
            listRelease(copy);
            listReleaseIterator(iter);
            return NULL;
        }
    }
    listReleaseIterator(iter);  //自行釋放迭代器
    return copy;    //返回拷貝副本
}

/* Search the list for a node matching a given key.
 * The match is performed using the 'match' method
 * set with listSetMatchMethod(). If no 'match' method
 * is set, the 'value' pointer of every node is directly
 * compared with the 'key' pointer.
 *
 * On success the first matching node pointer is returned
 * (search starts from head). If no matching node exists
 * NULL is returned. */
listNode *listSearchKey(list *list, void *key)  //在list中查詢value為key的節點並返回
{
    listIter *iter;
    listNode *node;

    iter = listGetIterator(list, AL_START_HEAD);    //建立迭代器
    while((node = listNext(iter)) != NULL) {        //迭代整個連結串列
        if (list->match) {                          //如果設定list結構中的match方法，則用該方法比較
            if (list->match(node->value, key)) {
                listReleaseIterator(iter);          //如果找到，釋放迭代器返回node地址
                return node;
            }
        } else {
            if (key == node->value) {
                listReleaseIterator(iter);
                return node;
            }
        }
    }
    listReleaseIterator(iter);      //釋放迭代器
    return NULL;
}

/* Return the element at the specified zero-based index
 * where 0 is the head, 1 is the element next to head
 * and so on. Negative integers are used in order to count
 * from the tail, -1 is the last element, -2 the penultimate
 * and so on. If the index is out of range NULL is returned. */
listNode *listIndex(list *list, long index) {   //返回下標為index的節點地址
    listNode *n;

    if (index < 0) {
        index = (-index)-1;         //如果下標為負數，從連結串列尾部開始
        n = list->tail;
        while(index-- && n) n = n->prev;
    } else {
        n = list->head;             //如果下標為正數，從連結串列頭部開始
        while(index-- && n) n = n->next;
    }
    return n;
}

/* Rotate the list removing the tail node and inserting it to the head. */
void listRotate(list *list) {       //將尾節點插到頭結點
    listNode *tail = list->tail;

    if (listLength(list) <= 1) return;  //只有一個節點或空連結串列直接返回

    /* Detach current tail */
    list->tail = tail->prev;        //取出尾節點，更新list的tail指標
    list->tail->next = NULL;
    /* Move it as head */
    list->head->prev = tail;        //將節點插到表頭，更新list的head指標
    tail->prev = NULL;
    tail->next = list->head;
    list->head = tail;
}

參考書籍：《Redis設計與實現》——黃健巨集