1. 程式人生 > 程式設計 >Redis原始碼解析-基礎資料-ziplist(壓縮列表)

Redis原始碼解析-基礎資料-ziplist(壓縮列表)

太長不看版

  • 壓縮列表是一種為節約空間而實現的線性資料結構,本質上是位元組陣列。
  • 壓縮列表元素可以為整數或字串。
  • 壓縮列表在快速列表、列表物件和雜湊物件中都有使用。
  • 壓縮列表新增(平均複雜度O(n))與刪除節點(平均複雜度O(n)),可能會觸發連鎖更新(平均複雜度O(n^2)),因為觸發機率不高所以不影響效能。
  • 因為節點儲存資料可能為字串,而字串匹配為O(n)複雜度,所以壓縮列表查詢節點平均複雜度為O(n^2)。

本篇解析基於redis 5.0.0版本,本篇涉及原始碼檔案為ziplist.c,ziplist.h。

什麼是壓縮列表

/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;
    unsigned char *zl = zmalloc(bytes);
    // ...
    return
zl; } 複製程式碼

從上述建立程式碼中可以看出,壓縮列表本質上就是一個位元組陣列,其是redis使用位元組陣列實現的線性資料結構,其元素可以是整數或者字串。在快速列表、列表物件和雜湊物件中都有使用,一般用在元素較少且元素位元組數較小的情況下。

為什麼要實現壓縮列表?

因為為了儘可能的節約記憶體。設想下現在有兩個元素: 整數1和字串'abc',兩個節點 資料本身只有佔8個位元組(64位機器中)。 使用單向連結串列的話需要多加兩個向後指標,佔用24個位元組,其中連結串列節點資料佔用16個位元組是資料本身大小的兩倍。 而此例中壓縮列表儲存只需要每個節點記錄前一節點長度(1位元組)、節點本身編碼資訊(1位元組)(具體規則後邊解釋

),總共佔用10個位元組。

壓縮列表的構成

#define ZIP_END 255         /* Special "end of ziplist" entry. */
#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl)))
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))
#define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))
#define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))
#define ZIPLIST_END_SIZE (sizeof(uint8_t)) /* Create a new empty ziplist. */ unsigned char *ziplistNew(void) { unsigned int bytes = ZIPLIST_HEADER_SIZE+1; unsigned char *zl = zmalloc(bytes); // 壓縮列表總位元組長度 ZIPLIST_BYTES(zl) = intrev32ifbe(bytes); // 尾部節點位元組距離 ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE); // 壓縮列表節點個數 ZIPLIST_LENGTH(zl) = 0; // 255特殊結尾值 zl[bytes-1] = ZIP_END; return zl; } 複製程式碼

壓縮列表由總位元組長度(4位元組),尾節點偏移量(4位元組),節點數量(2位元組),節點以及值為255的特殊結束符(1位元組)組成,通過列表的開始地址向後偏移尾節點偏移量個位元組,可以以O(1)時間複雜度獲取尾節點資訊。

壓縮列表自身的資訊只佔用了11個位元組,而連結串列光是頭指標和尾指標儲存就需要16個位元組,所以針對資料量少的情況(節點少且節點小)採用壓縮列表會比較划算。

intrev32ifbe函式為大小端轉換函式,統一轉換為小端儲存。為什麼要進行轉換?

因為壓縮列表的操作中涉及到的位運算很多,如果不統一的話會出現混亂。後續的所有位運算都是在小端儲存的基礎上進行的。大小端知識點戳此瞭解

壓縮列表節點的構成

typedef struct zlentry {
    // 前一節點長度資訊的長度
    unsigned int prevrawlensize;
    // 前一節點長度
    unsigned int prevrawlen;
    // 當前節點長度資訊長度
    unsigned int lensize;  
    // 當前節點長度
    unsigned int len;
    // 當前節點頭部資訊長度
    unsigned int headersize;
    // 當前節點資料編碼
    unsigned char encoding;     
    unsigned char *p;           
} zlentry;

void zipEntry(unsigned char *p,zlentry *e) {
    // 前一節點長度資訊解析
    ZIP_DECODE_PREVLEN(p,e->prevrawlensize,e->prevrawlen);
    // 當前節點資料長度與編碼資訊解析
    ZIP_DECODE_LENGTH(p + e->prevrawlensize,e->encoding,e->lensize,e->len);
    e->headersize = e->prevrawlensize + e->lensize;
    e->p = p;
}
複製程式碼

如前所述,壓縮列表本質上是位元組陣列,redis為了操作計算方便定義了zlentry結構體。進行操作計算時,將位元組陣列中包含的資訊按照規則解析到zlentry結構體中,方便後續的計算。從儲存角度來看,壓縮列表節點分為三部分:

  • 前一節點長度資訊
  • 當前節點資料長度與編碼資訊
  • 節點資料

下圖為壓縮列表節點各部分組成與zlentry結構體長度欄位示意圖:

前一節點長度資訊

#define ZIP_BIG_PREVLEN 254 
#define ZIP_DECODE_PREVLENSIZE(ptr,prevlensize) do {                          \
    if ((ptr)[0] < ZIP_BIG_PREVLEN) {                                          \
        (prevlensize) = 1;                                                     \
    } else {                                                                   \
        (prevlensize) = 5;                                                     \
    }                                                                          \
} while(0);

#define ZIP_DECODE_PREVLEN(ptr,prevlensize,prevlen) do {                     \
    ZIP_DECODE_PREVLENSIZE(ptr,prevlensize);                                  \
    if ((prevlensize) == 1) {                                                  \
        (prevlen) = (ptr)[0];                                                  \
    } else if ((prevlensize) == 5) {                                           \
        assert(sizeof((prevlen)) == 4);                                    \
        memcpy(&(prevlen),((char*)(ptr)) + 1,4);                             \
        memrev32ifbe(&prevlen);                                                \
    }                                                                          \
} while(0);
複製程式碼

前一節點長度小於254時,使用1個位元組儲存前一節點的長度資訊。

前一節點長度大於254時,使用5個位元組儲存前一節點的長度資訊。首個位元組固定為254,後續的4個位元組用來儲存長度資訊。

巨集定義為什麼要寫成do {} while(0);

寫成do {} while(0); 是為了保證不管在呼叫程式碼中怎麼使用分號和大括號,該巨集總是能確保其行為是一致的。錯誤示例如下:

#define TEST() test1(); test2();
// 此時test2必然被執行,不符合預期
if (a > 0)
    TEST();
#define TEST() { test1(); test2(); }
// 此時就會出現語法錯誤
if (a > 0)
    TEST();
複製程式碼

當前節點資料長度與編碼資訊

/* Different encoding/length possibilities */
#define ZIP_STR_MASK 0xc0         /* 11000000 */
#define ZIP_STR_06B (0 << 6)      /* 00000000 */
#define ZIP_STR_14B (1 << 6)      /* 01000000 */
#define ZIP_STR_32B (2 << 6)      /* 10000000 */
#define ZIP_INT_16B (0xc0 | 0<<4) /* 11000000 */
#define ZIP_INT_32B (0xc0 | 1<<4) /* 11010000 */
#define ZIP_INT_64B (0xc0 | 2<<4) /* 11100000 */
#define ZIP_INT_24B (0xc0 | 3<<4) /* 11110000 */
#define ZIP_INT_8B 0xfe           /* 11111110 */

/* 4 bit integer immediate encoding |1111xxxx| with xxxx between
 * 0001 and 1101. */
#define ZIP_INT_IMM_MASK 0x0f   /* 00001111 */
#define ZIP_INT_IMM_MIN 0xf1    /* 11110001 */
#define ZIP_INT_IMM_MAX 0xfd    /* 11111101 */

#define ZIP_ENTRY_ENCODING(ptr,encoding) do {  \
    (encoding) = (ptr[0]); \
    if ((encoding) < ZIP_STR_MASK) (encoding) &= ZIP_STR_MASK; \
} while(0)

#define ZIP_DECODE_LENGTH(ptr,encoding,lensize,len) do {                    \
    ZIP_ENTRY_ENCODING((ptr),(encoding));                                     \
    if ((encoding) < ZIP_STR_MASK) {                                           \
        if ((encoding) == ZIP_STR_06B) {                                       \
            (lensize) = 1;                                                     \
            (len) = (ptr)[0] & 0x3f;                                           \
        } else if ((encoding) == ZIP_STR_14B) {                                \
            (lensize) = 2;                                                     \
            (len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1];                       \
        } else if ((encoding) == ZIP_STR_32B) {                                \
            (lensize) = 5;                                                     \
            (len) = ((ptr)[1] << 24) |                                         \
                    ((ptr)[2] << 16) |                                         \
                    ((ptr)[3] <<  8) |                                         \
                    ((ptr)[4]);                                                \
        } else {                                                               \
            panic("Invalid string encoding 0x%02X",(encoding));               \
        }                                                                      \
    } else {                                                                   \
        (lensize) = 1;                                                         \
        (len) = zipIntSize(encoding);                                          \
    }                                                                          \
} while(0);

unsigned int zipIntSize(unsigned char encoding) {
    switch(encoding) {
    case ZIP_INT_8B:  return 1;
    case ZIP_INT_16B: return 2;
    case ZIP_INT_24B: return 3;
    case ZIP_INT_32B: return 4;
    case ZIP_INT_64B: return 8;
    }
    if (encoding >= ZIP_INT_IMM_MIN && encoding <= ZIP_INT_IMM_MAX)
        return 0; /* 4 bit immediate */
    panic("Invalid integer encoding 0x%02X",encoding);
    return 0;
}

int zipTryEncoding(unsigned char *entry,unsigned int entrylen,long long *v,unsigned char *encoding) {
    long long value;
    if (entrylen >= 32 || entrylen == 0) return 0;
    if (string2ll((char*)entry,entrylen,&value)) {
        if (value >= 0 && value <= 12) {
            *encoding = ZIP_INT_IMM_MIN+value;
        // ...
        return 1;
    }
    return 0;
}
複製程式碼

編碼與對應資料長度如下表所示:

編碼 長度資訊的長度 資料長度
ZIP_STR_06B(00bbbbbb)(前兩位標記編碼,後6位儲存長度) 1個位元組 長度<=63的位元組陣列
ZIP_STR_14B(01bbbbbb xxxxxxxx)(前2位為編碼後14位為長度) 2個位元組 長度<=16383的位元組陣列
ZIP_STR_06B(10______ aaaaaaaa bbbbbbbb cccccccc)(前2位標記編碼之後6位留空,後24位為長度) 5位元組 長度<=4294967295的位元組陣列
ZIP_INT_16B(11000000) 2位元組 int16_t(short)型別整數
ZIP_INT_32B(11010000) 4位元組 int32_t(int)型別整數
ZIP_INT_64B(11100000) 8位元組 int64_t(long long)型別整數
ZIP_INT_24B(11110000) 3位元組 3位元組長的有符號整數
ZIP_INT_8B(11111110) 1位元組 1位元組長的有符號整數
1111 xxxx 4位 0-12的無符號整數(此時該節點沒有content部分,資料儲存在encoding部分)

按照資料的型別和大小,匹配不同的編碼,思路和之前分析過的基礎型別sds使用5中頭部資訊一樣,都是為了最大化壓縮佔用空間。

壓縮列表相關操作

更新節點

#define ZIPLIST_HEAD 0
#define ZIPLIST_TAIL 1
#define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))
#define ZIPLIST_ENTRY_HEAD(zl)  ((zl)+ZIPLIST_HEADER_SIZE)
#define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)
unsigned char *ziplistPush(unsigned char *zl,unsigned char *s,unsigned int slen,int where) {
    unsigned char *p;
    // 頭部或尾部插入節點
    p = (where == ZIPLIST_HEAD) ? ZIPLIST_ENTRY_HEAD(zl) : ZIPLIST_ENTRY_END(zl);
    return __ziplistInsert(zl,p,s,slen);
}

unsigned char *ziplistInsert(unsigned char *zl,unsigned char *p,unsigned int slen) {
    /// 插入節點至p指標指向元素前方
    return __ziplistInsert(zl,slen);
}

unsigned char *__ziplistInsert(unsigned char *zl,unsigned int slen) {
    size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)),reqlen;
    unsigned int prevlensize,prevlen = 0;
    size_t offset;
    int nextdiff = 0;
    unsigned char encoding = 0;
    long long value = 123456789; 

    /* Find out prevlen for the entry that is inserted. */
    if (p[0] != ZIP_END) {
        // 插入位置不在尾部
        ZIP_DECODE_PREVLEN(p,prevlen);
    } else {
        // 插入位置在尾部
        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
        // 尾節點存在取長度,不存在長度為0(列表為空時不存在)
        if (ptail[0] != ZIP_END) {
            prevlen = zipRawEntryLength(ptail);
        }
    }
    // s指標指向新增節點資料 slen為資料長度
    // 確定資料編碼。資料長度,為整數時返回對應固定長度,為字串使用slen
    if (zipTryEncoding(s,slen,&value,&encoding)) {
        reqlen = zipIntSize(encoding);
    } else {
        reqlen = slen;
    }
    // 加上前一節點長度資訊的長度
    reqlen += zipStorePrevEntryLength(NULL,prevlen);
    // 加上編碼與長度資訊的長度
    // 此時reqlen為新加入節點的整體長度
    reqlen += zipStoreEntryEncoding(NULL,slen);

    int forcelarge = 0;
    // 
    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
    // 修復bug,詳細分析見:https://segmentfault.com/a/1190000018878466?utm_source=tag-newest
    if (nextdiff == -4 && reqlen < 4) {
        nextdiff = 0;
        forcelarge = 1;
    }

    offset = p-zl;
    // 調整記憶體大小
    zl = ziplistResize(zl,curlen+reqlen+nextdiff);
    p = zl+offset;

    // 非空列表插入
    if (p[0] != ZIP_END) {
        // 將p節點後移(沒有移動p節點前一節點長度資訊),留出當前節點位置
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);

        // 寫入p節點前一節點長度資訊(要插入節點的長度)
        if (forcelarge)
            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
        else
            zipStorePrevEntryLength(p+reqlen,reqlen);

        // 更新尾節點偏移量
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
        zipEntry(p+reqlen,&tail);
        if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
        }
    } else {
        // 空列表插入,只更新尾節點偏移量
        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
    }

    // 連鎖更新
    if (nextdiff != 0) {
        offset = p-zl;
        zl = __ziplistCascadeUpdate(zl,p+reqlen);
        p = zl+offset;
    }

    // 寫入前一節點長度資訊
    p += zipStorePrevEntryLength(p,prevlen);
    // 寫入節點編碼與長度資訊
    p += zipStoreEntryEncoding(p,slen);
    // 寫入資料
    if (ZIP_IS_STR(encoding)) {
        memcpy(p,slen);
    } else {
        zipSaveInteger(p,value,encoding);
    }
    // 增加列表長度
    ZIPLIST_INCR_LENGTH(zl,1);
    return zl;
}
複製程式碼

連鎖更新

unsigned char *__ziplistCascadeUpdate(unsigned char *zl,unsigned char *p) {
    size_t curlen = intrev32ifbe(ZIPLIST_BYTES(zl)),rawlen,rawlensize;
    size_t offset,noffset,extra;
    unsigned char *np;
    zlentry cur,next;

    while (p[0] != ZIP_END) {
        // 解析當前節點資訊
        zipEntry(p,&cur);
        // 當前節點總長
        rawlen = cur.headersize + cur.len;
        // 儲存當前節點長度資訊所需長度
        rawlensize = zipStorePrevEntryLength(NULL,rawlen);

        // 列表末尾,停止遍歷
        if (p[rawlen] == ZIP_END) break;
        // 解析下一節點資訊
        zipEntry(p+rawlen,&next);

        /* Abort when "prevlen" has not changed. */
        if (next.prevrawlen == rawlen) break;

        if (next.prevrawlensize < rawlensize) {
            /* The "prevlen" field of "next" needs more bytes to hold
             * the raw length of "cur". */
            offset = p-zl;
            // 下一節點因 前一節點長度資訊 欄位長度變更引發的自身長度變化大小
            extra = rawlensize-next.prevrawlensize;
            // 記憶體重新分配
            zl = ziplistResize(zl,curlen+extra);
            p = zl+offset;

            /* Current pointer and offset for next element. */
            np = p+rawlen;
            noffset = np-zl;

            // 如果下一節點不是尾節點,則需要更新 尾部節點偏移量
            if ((zl+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) != np) {
                ZIPLIST_TAIL_OFFSET(zl) =
                    intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra);
            }

            /* Move the tail to the back. */
            memmove(np+rawlensize,np+next.prevrawlensize,curlen-noffset-next.prevrawlensize-1);
            zipStorePrevEntryLength(np,rawlen);


            p += rawlen;
            curlen += extra;
        } else {
            // 如果 next節點原本的 前一節點長度資訊 欄位長度可以容納新插入節點的長度資訊,則直接寫入並退出遍歷
            if (next.prevrawlensize > rawlensize) {
                /* This would result in shrinking,which we want to avoid.
                 * So,set "rawlen" in the available bytes. */
                zipStorePrevEntryLengthLarge(p+rawlen,rawlen);
            } else {
                zipStorePrevEntryLength(p+rawlen,rawlen);
            }

            /* Stop here,as the raw length of "next" has not changed. */
            break;
        }
    }
    return zl;
}
複製程式碼

前邊我們看到了壓縮列表節點的各個欄位都是變長的,如果前一節點長度發生變化,可能會引起後一節點的長度發生變化(如果前一節點的長度之前小於254,變更後大於254,則後一節點的 前一節點長度資訊佔用位元組會從1個位元組變為5個位元組)。考慮一種極端情況,插入節點後續的節點長度都是介於250~253之間,此時插入一個節點長度大於254的節點,會引發連鎖更新,如下圖所示:

圖中每行都代表了一次記憶體重新分配,紅框框出的是每次擴張的位元組。此時插入節點的時間複雜度為O(n^2)(n 次空間重分配操作,每次空間重分配的最壞複雜度為 O(n) )。

連鎖更新最壞情況下帶來的效能消耗是災難性的,為什麼還可以放心的使用?

  • 觸發的條件(恰好存在多個連續的長度介於250~253之間的節點 )使得其觸發的概率很低
  • 壓縮列表的應用場景(節點數量較少且節點資料長度較小)以及長度恰好介於250-~253之間的情況出現情況較低使得即便出現了連鎖更新,需要更新的節點也沒有幾個。

所以壓縮列表插入節點的平均複雜度為O(n)。

查詢節點

unsigned char *ziplistFind(unsigned char *p,unsigned char *vstr,unsigned int vlen,unsigned int skip) {
    int skipcnt = 0;
    unsigned char vencoding = 0;
    long long vll = 0;
    // 遍歷壓縮列表 skip為查詢前跳過skip個節點
    while (p[0] != ZIP_END) {
        unsigned int prevlensize,len;
        unsigned char *q;

        ZIP_DECODE_PREVLENSIZE(p,prevlensize);
        ZIP_DECODE_LENGTH(p + prevlensize,len);
        q = p + prevlensize + lensize;

        if (skipcnt == 0) {
            /* Compare current entry with specified entry */
            if (ZIP_IS_STR(encoding)) {
                // 資料為字串,判斷字串相等複雜度為O(n)
                if (len == vlen && memcmp(q,vstr,vlen) == 0) {
                    return p;
                }
            } else {
                if (vencoding == 0) {
                    // 首次比對時,對傳入值進行解碼
                    if (!zipTryEncoding(vstr,vlen,&vll,&vencoding)) {
                        /* If the entry can't be encoded we set it to
                         * UCHAR_MAX so that we don't retry again the next
                         * time. */
                        vencoding = UCHAR_MAX;
                    }
                    /* Must be non-zero by now */
                    assert(vencoding);
                }

                /* Compare current entry with specified entry,do it only
                 * if vencoding != UCHAR_MAX because if there is no encoding
                 * possible for the field it can't be a valid integer. */
                if (vencoding != UCHAR_MAX) {
                    long long ll = zipLoadInteger(q,encoding);
                    if (ll == vll) {
                        return p;
                    }
                }
            }

            /* Reset skip count */
            skipcnt = skip;
        } else {
            /* Skip entry */
            skipcnt--;
        }

        /* Move to next entry */
        p = q + len;
    }

    return NULL;
}
複製程式碼

因為節點中存在字串,且字串相等比對的複雜度為O(n),所以壓縮列表查詢節點的複雜度為O(n^2)。

刪除節點

unsigned char *ziplistDelete(unsigned char *zl,unsigned char **p) {
    size_t offset = *p-zl;
    zl = __ziplistDelete(zl,*p,1);

    /* Store pointer to current element in p,because ziplistDelete will
     * do a realloc which might result in a different "zl"-pointer.
     * When the delete direction is back to front,we might delete the last
     * entry and end up with "p" pointing to ZIP_END,so check this. */
    *p = zl+offset;
    return zl;
}

/* Delete a range of entries from the ziplist. */
unsigned char *ziplistDeleteRange(unsigned char *zl,int index,unsigned int num) {
    unsigned char *p = ziplistIndex(zl,index);
    return (p == NULL) ? zl : __ziplistDelete(zl,num);
}

unsigned char *__ziplistDelete(unsigned char *zl,unsigned int num) {
    unsigned int i,totlen,deleted = 0;
    size_t offset;
    int nextdiff = 0;
    zlentry first,tail;
    // 刪除的首個節點
    zipEntry(p,&first);
    for (i = 0; p[0] != ZIP_END && i < num; i++) {
        // 跳過所有要刪除的節點
        p += zipRawEntryLength(p);
        deleted++;
    }
    // first.p指向第一個要被刪除節點的首地址
    // p指向的是最後一個刪除節點的下一節點的首地址,totlen大於0表示需要刪除節點
    totlen = p-first.p; /* Bytes taken by the element(s) to delete. */
    if (totlen > 0) {
        if (p[0] != ZIP_END) {
            // 獲取p節點 【前一節點長度資訊】 欄位與首個被刪除節點 【前一個節點長度資訊】 欄位的差值
            // zipPrevLenByteDiff 的返回值有三種可能:
            // 1)新舊兩個節點的【前一個節點長度資訊】欄位長度 相等,返回 0
            // 2)新節點【前一個節點長度資訊】欄位長度 > 舊節點【前一個節點長度資訊】欄位長度,返回 5 - 1 = 4
            // 3)舊節點【前一個節點長度資訊】欄位長度 > 新節點【前一個節點長度資訊】欄位長度,返回 1 - 5 = -4
            nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
            // 根據差值,向前向後偏移 p指標
            p -= nextdiff;
            // 將首個被刪除節點【前一節點長度資訊】寫入p指標指向的節點
            zipStorePrevEntryLength(p,first.prevrawlen);

            // 更新尾節點偏移量
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);

            /* When the tail contains more than one entry,we need to take
             * "nextdiff" in account as well. Otherwise,a change in the
             * size of prevlen doesn't have an effect on the *tail* offset. */
            zipEntry(p,&tail);
            /* 如果p節點不是尾節點,則尾節點偏移量需要加上nextdiff的變更量
               因為尾節點偏移量是指列表首地址到尾節點首地址的距離
               p節點的 【前一節點長度資訊】 欄位的長度變化隻影響它欄位之後的資訊地址。
               p節點為尾節點時,為節點首地址在【前一節點長度資訊】欄位前邊,所以不受影響。*/
            if (p[tail.headersize+tail.len] != ZIP_END) {
                ZIPLIST_TAIL_OFFSET(zl) =
                   intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
            }

            /* Move tail to the front of the ziplist */
            memmove(first.p,intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);
        } else {
            /* The entire tail was deleted. No need to move memory. */
            // 一直刪除到尾節點,不需要變更中間節點,只需要調整下尾節點偏移量
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe((first.p-zl)-first.prevrawlen);
        }

        /* Resize and update length */
        offset = first.p-zl;
        // 重新分配記憶體大小
        zl = ziplistResize(zl,intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
        // 減少節點長度
        ZIPLIST_INCR_LENGTH(zl,-deleted);
        p = zl+offset;

        // 如果最後一個被刪除節點的下一節點的【前一個節點長度資訊】欄位長度 需要變更,則可能會觸發連鎖更新
        if (nextdiff != 0)
            zl = __ziplistCascadeUpdate(zl,p);
    }
    return zl;
}
複製程式碼

因為可能會觸發連鎖更新,所以刪除操作最壞複雜度為O(n^2),平均複雜度為O(n)。