Redis源码阅读笔记(5)-压缩列表

压缩列表是做为列表键和哈希键的底层实现，是由一整块特殊编码的内存块所组成，非常适合放置少量列表项，并且每个列表项都是小整数值或者是短字符串。不得不佩服，Redis真是把内存使用到了极致。

涉及的主要代码：

1. ziplist.c
2. ziplist.h

ziplist的组成

ziplist

/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
    unsigned int bytes = ZIPLIST_HEADER_SIZE+1;     //1表示尾部
    unsigned char *zl = zmalloc(bytes);
    ZIPLIST_BYTES(zl) = intrev32ifbe(bytes);        //记录整个列表占用字节数
    ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(ZIPLIST_HEADER_SIZE);   //表尾节点到压缩列表起始有多个字节
    ZIPLIST_LENGTH(zl) = 0;     //当前长度为0
    zl[bytes-1] = ZIP_END;      //末端
    return zl;
}

属性	长度（字节）	描述
zlbytes	4	记录了ziplist所占的字节数
zltail	4	记录了ziplist头部节点到尾部节点的字节数，可以用来计算尾节点的地址
zllen	2	记录ziplist的节点数量；zllen=65535，表示节点数量需要遍历ziplist来获取
zlentry	-	节点，具体字节数由节点内容决定
zlend	1	用0xFF来表示节点末端

ziplist没有定义专门的结构体，其在内存块中的表示如图所示，其各个属性值如表格所示，其中zlentry可以是一个整数，也可以是一个字符串；一个压缩列表可以包含任意多（理论上）的节点。

/* 再内存中并没有存下下面结构体zlentry的内容，只是为了方便，在代码中定义了这样的结构体，包含了一些其他信息
*/
typedef struct zlentry {
    unsigned int prevrawlensize, prevrawlen;            //用来计算前面节点的地址
    unsigned int lensize, len;                          //本节点的长度
    unsigned int headersize;                            //头部大小
    unsigned char encoding;                             //编码
    unsigned char *p;
} zlentry;

zlentry

z'len't'r'y

zlentry由以下3各部分组成：

1. prevrawlen：记录前一个节点所占有的内存字节数，通过该值，我们可以从当前节点计算前一个节点的地址，可以用来实现表尾向表头节点遍历；
2. len/encoding：记录了当前节点content占有的内存字节数及其存储类型，用来解析content用；
3. content：保存了当前节点的值。

最关键的是prevrawlen和len/encoding，content只是实际存储数值的比特位。

prevrawlen

prevrawlen

prevrawlen是变长编码，有两种表示方法：

1. 如果长度小于254字节，则使用1字节(uint8_t)来存储prevrawlen；
2. 如果长度大于或等于254字节，则使用4字节(uint32_t)来存储prevrawlen。

因为oxFF已经用来做zipend，因此使用0xFE(254)来作为标识符区分。

#define ZIP_BIGLEN 254

//前节点长度写入
//p如果是NULL，则只计算长度，否则，要写入p
static unsigned int zipPrevEncodeLength(unsigned char *p, unsigned int len) {
    if (p == NULL) {
        return (len < ZIP_BIGLEN) ? 1 : sizeof(len)+1;
    } else {
        if (len < ZIP_BIGLEN) {
            p[0] = len;
            return 1;
        } else {                //大于254的长度，头字节为 0xfe
            p[0] = ZIP_BIGLEN;
            memcpy(p+1,&len,sizeof(len));           //写入后面4个字节
            memrev32ifbe(p+1);
            return 1+sizeof(len);
        }
    }
}

//使用4个字节的长度来保存小于254的长度
static void zipPrevEncodeLengthForceLarge(unsigned char *p, unsigned int len) {
    if (p == NULL) return;
    p[0] = ZIP_BIGLEN;
    memcpy(p+1,&len,sizeof(len));
    memrev32ifbe(p+1);
}

//获取存储前字节长度的pre 字节长度
#define ZIP_DECODE_PREVLENSIZE(ptr, prevlensize) do {                          \
    if ((ptr)[0] < ZIP_BIGLEN) {                                               \
        (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((prevlensize)) == 4);                                    \
        memcpy(&(prevlen), ((char*)(ptr)) + 1, 4);                             \
        memrev32ifbe(&prevlen);                                                \
    }                                                                          \
} while(0);

len/encoding

编码是压缩列表中比较难的地方，只要理解如何编码，如何解码就水到渠成

len/encoding

压缩表支持以下几种类型：字符串、4位长无符号整数、int8_t、int16_t、3字节长有符号整数、int32_t、int64_t。

字符串：

1. 00aa aaaa： 00表示编码和长度使用1个字节表示，剩余的6位比特位用来表示具体的字节长度；
2. 01aa aaaa aaaa aaaa：01表示编码和长度使用2个字节表示，剩余的14位比特用来表示具体的字节长度；
3. 10__ ____ aaaa aaaa aaaa aaaa aaaa aaaa aaaa aaaa：10用来表示编码和长度使用5个字节，10接下来的6位比特不使用，再接下来的4个字节用来表示具体的字节长度。

整数：使用11开头的数据来表示

1. 1100 0000：表示int16_t；
2. 1101 0000：表示int32_t；
3. 1110 0000：表示int64_t；
4. 1111 0000：3字节长有符号整数；
5. 1111 1110：表示in8t_t；
6. 1111 bbbb：该项比较特殊，编码和content是放在一起，该项bbbb表示实际的数据项，由于0xff与zipend冲突，0xfe与int8_t编码冲突，ox10与3字节有符号整数冲突，因此0xff/0xfe/0x00均不使用，而最小的值为0，最大的值为12，因此实际的值应该是convertBinaryToDecimal(bbbb) - 1

/* Different encoding/length possibilities */
#define ZIP_STR_MASK 0xc0
#define ZIP_STR_06B (0 << 6)        //00bbbbbb    1字节长     2^6
#define ZIP_STR_14B (1 << 6)        //01bbbbbb xxxxxxxx   2字节长 2^14
#define ZIP_STR_32B (2 << 6)        //10______ aaaaaaaa bbbbbbbb cccccccc dddddddd    5字节长 2^32   
#define ZIP_INT_16B (0xc0 | 0<<4)   //11000000 1字节
#define ZIP_INT_32B (0xc0 | 1<<4)   //11010000 1字节
#define ZIP_INT_64B (0xc0 | 2<<4)   //11100000 1字节
#define ZIP_INT_24B (0xc0 | 3<<4)   //11110000 1字节    
#define ZIP_INT_8B 0xfe             //11111110 1字节   

#define ZIP_IS_STR(enc) (((enc) & ZIP_STR_MASK) < ZIP_STR_MASK)     //判断是否是字符串

//判断该数字字符串能否被编码以及编码类型
static int zipTryEncoding(unsigned char *entry, unsigned int entrylen, long long *v, unsigned char *encoding) {
    long long value;

    if (entrylen >= 32 || entrylen == 0) return 0;      //只能编码 占4字节的整数
    if (string2ll((char*)entry,entrylen,&value)) {
        /* Great, the string can be encoded. Check what's the smallest
         * of our encoding types that can hold this value. */
        if (value >= 0 && value <= 12) {
            *encoding = ZIP_INT_IMM_MIN+value;          //4位整数
        } else if (value >= INT8_MIN && value <= INT8_MAX) {
            *encoding = ZIP_INT_8B;                     //1字节整数
        } else if (value >= INT16_MIN && value <= INT16_MAX) {
            *encoding = ZIP_INT_16B;                    //2字节整数
        } else if (value >= INT24_MIN && value <= INT24_MAX) {
            *encoding = ZIP_INT_24B;                    //3字节
        } else if (value >= INT32_MIN && value <= INT32_MAX) {
            *encoding = ZIP_INT_32B;                    //4字节
        } else {
            *encoding = ZIP_INT_64B;                    //8字节
        }
        *v = value;
        return 1;
    }
    return 0;
}

//计算并存储字节编码
static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, unsigned int rawlen) {
    unsigned char len = 1, buf[5];

    if (ZIP_IS_STR(encoding)) {
        /* Although encoding is given it may not be set for strings,
         * so we determine it here using the raw length. */
        //通过字符串长度来判断编码类型
        if (rawlen <= 0x3f) {           //1字节长可表示        长度最大为 0011 1111()ox3f
            if (!p) return len;
            buf[0] = ZIP_STR_06B | rawlen;      //保存长度  
        } else if (rawlen <= 0x3fff) {          //长度最大为 0011 1111 1111 1111
            len += 1;
            if (!p) return len;
            buf[0] = ZIP_STR_14B | ((rawlen >> 8) & 0x3f);      //取前6位，然后组成的第一个字节  01开头
            buf[1] = rawlen & 0xff;         //组成第二个字节
        } else {
            len += 4;
            if (!p) return len;
            //计算各个字节怎么存
            buf[0] = ZIP_STR_32B;
            buf[1] = (rawlen >> 24) & 0xff;
            buf[2] = (rawlen >> 16) & 0xff;
            buf[3] = (rawlen >> 8) & 0xff;
            buf[4] = rawlen & 0xff;
        }
    } else {
        /* Implies integer encoding, so length is always 1. */
        //整数的编码字节都为1
        if (!p) return len;
        buf[0] = encoding;
    }

    /* Store this length at p */
    memcpy(p,buf,len);
    return len;
}

//解码字节编码，获取数据的字节数
#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) {    /*2字节长读的字符串*/                            \
            (lensize) = 2;                                                     \
            (len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1];                       \
        } else if (encoding == ZIP_STR_32B) {        /*5字节长度的字符串*/                           \
            (lensize) = 5;                                                     \
            (len) = ((ptr)[1] << 24) |                                         \
                    ((ptr)[2] << 16) |                                         \
                    ((ptr)[3] <<  8) |                                         \
                    ((ptr)[4]);                                                \
        } else {                                                               \
            assert(NULL);                                                      \
        }                                                                      \
    } else {                           /*否则是数字*/                                        \
        (lensize) = 1;                                                         \
        (len) = zipIntSize(encoding);                                          \
    }                                                                          \
} while(0);

//获取当前节点p存储的值
unsigned int ziplistGet(unsigned char *p, unsigned char **sstr, unsigned int *slen, long long *sval) {
    zlentry entry;
    if (p == NULL || p[0] == ZIP_END) return 0;
    if (sstr) *sstr = NULL;

    entry = zipEntry(p);
    if (ZIP_IS_STR(entry.encoding)) {
        if (sstr) {
            *slen = entry.len;
            *sstr = p+entry.headersize;
        }
    } else {
        if (sval) {
            *sval = zipLoadInteger(p+entry.headersize,entry.encoding);
        }
    }
    return 1;
}

//获得数字(content)
static int64_t zipLoadInteger(unsigned char *p, unsigned char encoding) {
    int16_t i16;
    int32_t i32;
    int64_t i64, ret = 0;
    if (encoding == ZIP_INT_8B) {
        ret = ((int8_t*)p)[0];
    } else if (encoding == ZIP_INT_16B) {
        memcpy(&i16,p,sizeof(i16));
        memrev16ifbe(&i16);
        ret = i16;
    } else if (encoding == ZIP_INT_32B) {
        memcpy(&i32,p,sizeof(i32));
        memrev32ifbe(&i32);
        ret = i32;
    } else if (encoding == ZIP_INT_24B) {
        i32 = 0;
        memcpy(((uint8_t*)&i32)+1,p,sizeof(i32)-sizeof(uint8_t));
        memrev32ifbe(&i32);
        ret = i32>>8;
    } else if (encoding == ZIP_INT_64B) {
        memcpy(&i64,p,sizeof(i64));
        memrev64ifbe(&i64);
        ret = i64;
    } else if (encoding >= ZIP_INT_IMM_MIN && encoding <= ZIP_INT_IMM_MAX) {
        ret = (encoding & ZIP_INT_IMM_MASK)-1;        //数值减 1，才能获取到实际值
    } else {
        assert(NULL);
    }
    return ret;
}

连锁更新

因为在ziplist中，每个zlentry都存储着前一个节点所占的字节数，而这个数值又是变长编码的。假设存在一个压缩列表，其包含e1、e2、e3、e4.....，e1节点的大小为253字节，那么e2.prevrawlen的大小为1字节，如果此时在e2与e1之间插入了一个新节点e_new，e_new编码后的整体长度（包含e1的长度）为254字节，此时e2.prevrawlen就需要扩充为5字节；如果e2的整体长度变化又引起了e3.prevrawlen的存储长度变化，那么e3也需要扩.......如此递归直到表尾节点或者某一个节点的prevrawlen本身长度可以容纳前一个节点的变化。其中每一次扩充都需要进行空间再分配操作。删除节点亦是如此，只要引起了操作节点之后的节点的prevrawlen的变化，都可能引起连锁更新。
连锁更新在最坏情况下需要进行N次空间再分配，而每次空间再分配的最坏时间复杂度为O(N)，因此连锁更新的总体时间复杂度是O(N^2)。
即使涉及连锁更新的时间复杂度这么高，但它能引起的性能问题的概率是极低的：需要列表中存在大量的节点长度接近254。

添加节点

1. 计算在当前位置插入，新节点所需要的大小；
2. 根据需要扩充的大小，扩大ziplist的空间；
3. 计算插入位置之后的节点是否需要变换prevrawlen的存储字节长度及其变量量(nextdff)；
4. 移动之后的节点到正确的位置；
5. 根据nextdiff判断是否需要连锁更新；
6. 设定当前节点的值；
7. 完成添加。

在改变后面节点的prevrawlen的值时候，Redis做了一个小改动，即当prevrawlen的字节长度需要变小的，Redis并改小存储字节长度，而是强迫使用大字节存储小数值。

//添加节点    
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);
}

//获取存储len所需字节和当前p地址存储的前节点长度所需字节的差异
static int zipPrevLenByteDiff(unsigned char *p, unsigned int len) {
    unsigned int prevlensize;
    ZIP_DECODE_PREVLENSIZE(p, prevlensize);
    return zipPrevEncodeLength(NULL, len) - prevlensize;
}

//重定义ziplist的大小
static unsigned char *ziplistResize(unsigned char *zl, unsigned int len) {
    zl = zrealloc(zl,len);
    ZIPLIST_BYTES(zl) = intrev32ifbe(len);
    zl[len-1] = ZIP_END;
    return zl;
}

//连锁更新，zl->ziplist的起始地址  p开始检测的地址
static 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) {
        cur = zipEntry(p);  //逐步获取每个节点
        rawlen = cur.headersize + cur.len;      //当前p节点的长度 
        rawlensize = zipPrevEncodeLength(NULL,rawlen);

        /* Abort if there is no next entry. */
        if (p[rawlen] == ZIP_END) break;
        next = zipEntry(p+rawlen);  //获取下一个节点

        /* 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;                              //重新计算p的地址，因为zl可能改变

            /* Current pointer and offset for next element. */
            np = p+rawlen;          //p节点之后的地址
            noffset = np-zl;        //p节点之后距离头部的距离

            /* Update tail offset when next element is not the tail element. */
            if ((zl+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))) != np) {
                ZIPLIST_TAIL_OFFSET(zl) =
                    intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+extra);     //修改zltail
            }

            /* Move the tail to the back. */
            memmove(np+rawlensize,      //从np的encoding往后移动
                np+next.prevrawlensize,
                curlen-noffset-next.prevrawlensize-1);
            zipPrevEncodeLength(np,rawlen);

            /* Advance the cursor */
            p += rawlen;
            curlen += extra;
        } else {    
            if (next.prevrawlensize > rawlensize) { //变小了
               //强制使用大字节存储小数，避免更新
                zipPrevEncodeLengthForceLarge(p+rawlen,rawlen);
            } else {            //不变
                zipPrevEncodeLength(p+rawlen,rawlen);       //改变值即可
            }

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

//在p处插入s
static unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, 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; /* initialized to avoid warning. Using a value
                                    that is easy to see if for some reason
                                    we use it uninitialized. */
    zlentry tail;

    if (p[0] != ZIP_END) {
        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);        //获取p除记录前节点长度的数据
    } else {
        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);
        if (ptail[0] != ZIP_END) {
            prevlen = zipRawEntryLength(ptail);                    //获取表尾节点的长度
        }
    }

    /* See if the entry can be encoded */
    if (zipTryEncoding(s,slen,&value,&encoding)) {          //尝试要对当前的值进行整数编码
        /* 'encoding' is set to the appropriate integer encoding */
        reqlen = zipIntSize(encoding);          //需要存储当前节点值需要的字节数
    } else {
        /* 'encoding' is untouched, however zipEncodeLength will use the
         * string length to figure out how to encode it. */
        reqlen = slen;
    }
 
    reqlen += zipPrevEncodeLength(NULL,prevlen);        //存储前节点长度需要的字节数
    reqlen += zipEncodeLength(NULL,encoding,slen);       //存储encoding需要的字节数

    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;    //看是否会引起后面节点的长度变化

    offset = p-zl;
    zl = ziplistResize(zl,curlen+reqlen+nextdiff);
    p = zl+offset;

    /* Apply memory move when necessary and update tail offset. */
    if (p[0] != ZIP_END) {
        /* 数据往后移动，并根据nextdiff进行微调 */  
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);    

        /* 改变原p处节点的prevrawlen */
        zipPrevEncodeLength(p+reqlen,reqlen);

        /* Update offset for tail */
        ZIPLIST_TAIL_OFFSET(zl) =
            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);

        /* 因为nextdiff，尾节点也需要微调*/
        tail = zipEntry(p+reqlen);
        if (p[reqlen+tail.headersize+tail.len] != ZIP_END) {
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
        }
    } else {
        /* This element will be the new tail. */
        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);
    }

    /* When nextdiff != 0, the raw length of the next entry has changed, so
     * we need to cascade the update throughout the ziplist */
    if (nextdiff != 0) {
        offset = p-zl;    //记录新节点的偏移量
        zl = _ziplistCascadeUpdate(zl,p+reqlen);    //连锁更新
        p = zl+offset;  //获取新节点的新地址
    }

    /* 编辑新节点的值 */
    p += zipPrevEncodeLength(p,prevlen);
    p += zipEncodeLength(p,encoding,slen);
    if (ZIP_IS_STR(encoding)) {
        memcpy(p,s,slen);
    } else {
        zipSaveInteger(p,value,encoding);
    }
    ZIPLIST_INCR_LENGTH(zl,1);
    return zl;
}

删除节点

1. 计算要删除的节点长度；
2. 计算最后一个删除节点之后的节点entry_a（如果有的话）的prevrawlen的存储长度是否需要改变及其变更量nextdiff；
3. 移动entry_a的起始地址（根据nextdiff）,重新设置该节点的起始地址，并设置其prevrawlen的值；
4. 移动entry_a及其之后的节点到合适的位置，并根据entry_a是否是未节点和nextdiff的值来调整ziptail的值；
5. 重新设置ziplist的空间；
6. 根据nextdiff的值看是否需要进行连锁更新。

//删除节点
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, unsigned int index, unsigned int num) {
    unsigned char *p = ziplistIndex(zl,index);
    return (p == NULL) ? zl : __ziplistDelete(zl,p,num);
}

//删除num个节点
static unsigned char *__ziplistDelete(unsigned char *zl, unsigned char *p, unsigned int num) {
    unsigned int i, totlen, deleted = 0;
    size_t offset;
    int nextdiff = 0;
    zlentry first, tail;

    first = zipEntry(p);
    for (i = 0; p[0] != ZIP_END && i < num; i++) {
        p += zipRawEntryLength(p);          //计算下一个节点的地址
        deleted++;
    }

    totlen = p-first.p;         //删除了多长
    if (totlen > 0) {
        //p已经是最后一个删除节点之后的节点
        if (p[0] != ZIP_END) {
            nextdiff = zipPrevLenByteDiff(p,first.prevrawlen); 
            //比较p的prerawlen和第一个被删除节点的prerawlen的区别（因为p的prevrawlen正好要更新为被第一个删除节点的prevrawlen正好）
            p -= nextdiff;
            zipPrevEncodeLength(p,first.prevrawlen);        //写入prevrawlen

            /* Update offset for tail */
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);

            /* 如果p是最后一个节点，那么zltail就保持不变，如果不是，因为有nextdiff的偏移，因此nextdiff也需要偏移 */
            tail = zipEntry(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,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;
}

获取列表长度

由于存储列表长度的是2字节，如果列表长度大于65535的将会溢出，因此Redis将0xffff作为表示长度>=65535，需要通过遍历列表来获取。

//返回压缩列表长度，如果小于UINT16_MAX,可以直接获取，否则要遍历计算
unsigned int ziplistLen(unsigned char *zl) {
    unsigned int len = 0;
    if (intrev16ifbe(ZIPLIST_LENGTH(zl)) < UINT16_MAX) {
        len = intrev16ifbe(ZIPLIST_LENGTH(zl));
    } else {
        unsigned char *p = zl+ZIPLIST_HEADER_SIZE;
        while (*p != ZIP_END) {
            p += zipRawEntryLength(p);
            len++;
        }

        /* Re-store length if small enough */
        if (len < UINT16_MAX) ZIPLIST_LENGTH(zl) = intrev16ifbe(len);
    }
    return len;
}

寻找节点

1. 如果当前节点是字符串，则比较字符串是否相等；
2. 如果当前节点是整数，将传入的字符串转化为整数再进行比较。

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;

    while (p[0] != ZIP_END) {
        unsigned int prevlensize, encoding, lensize, len;
        unsigned char *q;

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

        if (skipcnt == 0) {
            /* Compare current entry with specified entry */
            if (ZIP_IS_STR(encoding)) {
                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);
                }

                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;
}

总结

虽然ziplist相对其他类型代码量偏少，但其中所涉及的内容还是十分多的，通过不同的编码方式，使得ziplist可以用十分高效的方法存储整数和字符串，从而节省内存；并且由于**ziplist是内存连续，因此载入缓存速度将会很快。

ziplist主要API

function	description	time complexity
ziplistNew	创建一个新的压缩列表	O(1)
ziplistPush	创建一个新节点，并防止到ziplist的表头或表尾	平均O(N)，最坏O(N^2)
ziplistIndex	返回给定位置的节点	O(N)
ziplistnext		返回给定节点的下一个节点	O(1)
ziplistPrev	返回给定节点的前一个节点	O(1)
ziplistGet	返回给定节点的值	O(1)
ziplistInsert	创建一个节点，并放置到指定位置	平均O(N)，最坏O(N^2)
ziplistDelete	删除给定节点	平均O(N)，最坏O(N^2)
ziplistDeleteRange	删除连续多个节点	平均O(N)，最坏O(N^2)
ziplistFind	查找并返回节点	需要遍历列表并逐个比较，因此O(N^2)
ziplistLen	返回列表节点数量	数量小于0xfffe时O(1)，否则O(N)