redis源码分析之压缩链表ziplist

好久没写文章了，最近一直在看比特币方面的东西，虽然还没有了解的很深，但已经感觉到比特币系统设计的非常精致。研究比特币的知识也让我学到了计算机领域的一些基础知识，希望后续有机会跟大家分享比特币的技术。

言归正传，这篇文章总结一下redis中使用非常多的一个数据结构压缩链表ziplist。该数据结构在redis中说是无处不在也毫不过分，除了链表以外，很多其他数据结构也是用它进行过渡的，比如前面文章提到的SortedSet。

一、压缩链表ziplist数据结构简介

首先从整体上看下ziplist的结构，如下图：

压缩链表ziplist结构图

可以看出字段很多，字节大小也不同，但这也就是压缩链表的精髓所在了，我们依次总结一下。

字段	含义
zlbytes	该字段是压缩链表的第一个字段，是无符号整型，占用4个字节。用于表示整个压缩链表占用的字节数（包括它自己）。
zltail	无符号整型，占用4个字节。用于存储从压缩链表头部到最后一个entry（不是尾元素zlend）的偏移量，在快速跳转到链表尾部的场景使用。
zllen	无符号整型，占用2个字节。用于存储压缩链表中包含的entry总数。
zlend	特殊的entry，用来表示压缩链表的末尾。占用一个字节，值恒为255。

总结为ziplist的头跟尾，下面再总结一下重中之重的entry字段。

一般来说，一个entry由prevlen，encoding，entry-data三个字段组成，但当entry是个很小的整数时，会根据编码省略掉entry-data字段。下面依次进行总结：

首先是字段prevlen：表示前一个entry的长度，有两种编码方式。

1. 当长度小于255字节时，用一个字节存储。
2. 当长度大于等于255时，用五个字节进行存储，其中第一个字节会被设置为255表示前一个entry的长度由后面四个字节表示。

然后是字段encoding：它会根据当前元素内容的不同会采用不同的编码方式，如下：

1. 如果元素内容为字符串，encoding的值分别为：
   00xx xxxx ：00开头表示该字符串的长度用6个bit表示。
   01xx xxxx | xxxx xxxx ：01开头表示字符串的长度由14bit表示，这14个bit采用大端存储。
   1000 0000 | xxxx xxxx | xxxx xxxx | xxxx xxxx | xxxx xxxx ：10开头表示后续的四个字节为字符串长度，这32个bit采用大端存储。

2. 如果元素内容为数字，encoding的值分别为：
   1100 0000：表示数字占用后面2个字节。
   1101 0000：表示数字占用后面4个字节。
   1110 0000：表示数字占用后面8个字节。
   1111 0000 ：表示数字占用后面3个字节。
   1111 1110 ：表示数字占用后面1个字节。
   1111 1111 ：表示压缩链表中最后一个元素（特殊编码）。
   1111 xxxx ：表示只用后4位表示0~12的整数，由于0000，1110跟1111三种已经被占用，也就是说这里的xxxx四位只能表示0001~1101，转换成十进制就是数字1~13，但是redis规定它用来表示0~12，因此当遇到这个编码时，我们需要取出后四位然后减1来得到正确的值。

最后是字段entry-data：如果元素的值为字符串，则保存元素本身的值。如果元素的值为很小的数字（按上面编码规则即0~12），则没有该字段。

压缩链表的编码非常复杂，但这也正是该数据结构的精髓所在，一起来看一个例子吧：
注：这个例子是redis源码中提到的

//由元素2，5组成的压缩链表
[0f 00 00 00] [0c 00 00 00] [02 00] [00 f3] [02 f6] [ff]
      |             |          |       |       |     |
   zlbytes        zltail    entries   "2"     "5"   end

//字符串"Hello World"编码后的内容
[02] [0b] [48 65 6c 6c 6f 20 57 6f 72 6c 64]

上面是一段用十六进制表示2，5两个元素组成的压缩链表。

1. 首先前四个字节表示整个ziplist占用的字节数，因为redis采用小端存储，所以15个字节表示为0f 00 00 00。

2. 接下来的4个字节表示末尾元素偏移量，是从ziplist头（zlbytes）开始到最后一个元素（注：不是尾节点）的偏移量，也是采用小端存储，因此表示为0c 00 00 00。

3. 再后面是由两个字节组成的表示元素个数的zllen字段，在我们这个例子中有两个元素，加上小端存储，所以表示为02 00。

4. 接下来是元素本身，首先由一个变长的字端表示前一个元素的长度，2作为第一个元素，它前一个元素的大小就是0，因此占用一个字节00。按照我们上面说的编码规则，元素2跟5属于0~12之间的数字，需要使用1111 xxxx格式进行编码，2转成二进制为0010，再加上1就是0011（加1的原因上面已经说明），组合起来元素2就表示为00 f3。同理元素5表示为02 f6。

5. 最后就是尾元素，使用未被占用的编码1111 1111即255。

接下来我们在这个压缩链表末尾插入一个字符串元素Hello World，先看看该如何编码该字符串，按照约定的编码规则，首先要用字节表示前一个元素的长度，这里前一个元素时5，总共占用了两个字节，因此首先用一个字节表示前一个元素的长度02，接下来是字符串的编码，我们要加入的字符串长度为11（空格也算），转换成二进制就是1011，按照字符串的编码规则，使用0000 1011表示，转为十六进制就是0b，最后再加上我们字符串本身的ASCII码，综合起来就得出该字符串的编码：[02] [0b] [48 65 6c 6c 6f 20 57 6f 72 6c 64]。

此时整个压缩链表就变为：

[0f 00 00 00] [0c 00 00 00] [02 00] [00 f3] [02 f6] [02 0b 48 65 6c 6c 6f 20 57 6f 72 6c 64] [ff]
      |             |          |       |       |                        |                      |
   zlbytes        zltail    entries   "2"     "5"                 "Hello World"               end

二、压缩链表ziplist命令源码分析

搞明白上面的编码规则以后，我们再来看下压缩链表ziplist的部分操作源码，本文就创建压缩链表，插入元素，删除元素，查找元素四个操作来总结一下压缩链表的基本原理。

首先是创建：

//定义由zlbytes，zltail跟zllen组成的压缩链表的头大小
#define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))

//创建一个压缩链表，并且返回指向该链表的指针
unsigned char *ziplistNew(void) {
    //这里之所以+1是因为尾元素占用一个字节，这也是一个压缩链表最小尺寸
    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;
    //设置尾元素（上面只是申请空间）
    zl[bytes-1] = ZIP_END;
    return zl;
}

创建压缩链表的逻辑很简单，就是申请固定的包含头尾节点的空间，然后初始化链表上下文。

与创建相比，添加元素的源码就非常冗长了，为了便于理解，在看源码之前我们先自己梳理一下添加元素的逻辑。

1. 首先我们要找到指定插入位置的前一个元素的大小，因为该属性是新元素的组成部分之一。

2. 然后我们要对当前元素进行编码来获得相应的encoding字段跟实际元素值的字段。

3. 新插入元素的后继元素的prevlen字段要更新，因为它前面的元素已经改变。这里可能引起级联更新（删除元素也有该问题），原因就是prevlen字段大小是可变的。

上面三步是核心步骤，其余的还有更新尾节点偏移量，修改链表元素个数等操作。当然，由于压缩链表是基于数组实现的，因此在插入或删除元素的时候内存拷贝也是必不可少的。

总结好上面的步骤以后，我们开始一步一步分析源码，比较长，慢慢看：

//四个参数依次是：压缩链表，插入位置（新元素插入p元素后面），元素值，元素长度
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;
    zlentry tail;

    //1. 这段逻辑目的就是获取前置元素的长度
    if (p[0] != ZIP_END) {
        //如果插入位置的元素不是尾元素,则获取该元素的长度
        //这里为了后面使用方便进行了拆分，prevlensize保存encoding字段的长度，prevlen保存元素本身的长度
        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);
    } else {
        //如果插入位置的元素是尾元素，那么需要把新元素插入链表尾端
        //获取到链表最后一个元素（注：最后一个元素不等于尾元素）
        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);

        if (ptail[0] != ZIP_END) {
            //如果最后一个元素不是尾元素，则该元素为新元素的前置元素，获取该元素长度
            prevlen = zipRawEntryLength(ptail);
        }
        //否则说明链表还没有任何元素，即新元素的前置元素长度为0
    }

    //2. 对新元素进行编码，获取新元素的总大小
    if (zipTryEncoding(s,slen,&value,&encoding)) {
        //如果是数字，则按数字进行编码
        reqlen = zipIntSize(encoding);
    } else {
        //元素长度即为字符串长度
        reqlen = slen;
    }
    //新元素总长度为值的长度加上前置元素跟encoding元素的长度
    reqlen += zipStorePrevEntryLength(NULL,prevlen);
    reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

    //如果插入的位置不是链表尾，则要对新元素的后续元素的prevlen字段进行判断
    //根据上面的编码规则，该字段可能需要扩容
    int forcelarge = 0;
    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;
    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) {
        //把新元素后继元素复制到新的数组中，-1为尾元素
        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);
        //新元素的后继元素的prevlen字段
        if (forcelarge)
            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);
        else
            zipStorePrevEntryLength(p+reqlen,reqlen);

        //更新最后一个元素的偏移量
        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);
        //当新元素的后继元素不止有一个时，新的尾元素偏移量需要加上nextdiff
        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);
    }
    //nextdiff !=0表示后继元素的长度发生变化，因此我们需要级联更新后继元素的后继元素
    if (nextdiff != 0) {
        offset = p-zl;
        zl = __ziplistCascadeUpdate(zl,p+reqlen);
        p = zl+offset;
    }
    //把新元素写入链表
    p += zipStorePrevEntryLength(p,prevlen);
    p += zipStoreEntryEncoding(p,encoding,slen);
    if (ZIP_IS_STR(encoding)) {
        memcpy(p,s,slen);
    } else {
        zipSaveInteger(p,value,encoding);
    }
    //压缩链表存储元素数量+1
    ZIPLIST_INCR_LENGTH(zl,1);
    return zl;
}

分析完插入元素的逻辑，长舒一口气，真的很长，细节也很多。

接下来在再看下删除元素的过程，与添加相比，删除相对要简单一些，清空当前元素以后，需要把后继元素一个一个拷贝上来（这也是数组跟链表两个数据结构的差别），然后注意是否需要级联更新，上代码：

//参数依次为：压缩链表，删除元素的其实位置，删除元素的个数
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;
    //读取p指向的元素保存在first中
    zipEntry(p, &first);
    for (i = 0; p[0] != ZIP_END && i < num; i++) {
        //统计总共删除的长度
        p += zipRawEntryLength(p);
        //统计实际删除元素的个数
        deleted++;
    }
    //需要删除元素的字节数
    totlen = p-first.p;
    if (totlen > 0) {
        if (p[0] != ZIP_END) {
            //判断元素大小是否有改变
            nextdiff = zipPrevLenByteDiff(p,first.prevrawlen);
            //修改删除元素之后的元素的prevlen字段
            p -= nextdiff;
            zipStorePrevEntryLength(p,first.prevrawlen);

            //更新末尾元素的偏移量
            ZIPLIST_TAIL_OFFSET(zl) =intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))-totlen);
            //当删除元素的后继元素不止有一个时，新的末尾元素偏移量需要加上nextdiff
            zipEntry(p, &tail);
            if (p[tail.headersize+tail.len] != ZIP_END) {
                ZIPLIST_TAIL_OFFSET(zl) =
                   intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+nextdiff);
            }
            //把后面剩余的元素移动至前面
            memmove(first.p,p,
                intrev32ifbe(ZIPLIST_BYTES(zl))-(p-zl)-1);
        } else {
            //直接删除到链表末尾，因此不需要内存拷贝，只需修改最后一个元素的偏移量
            ZIPLIST_TAIL_OFFSET(zl) =
                intrev32ifbe((first.p-zl)-first.prevrawlen);
        }
        //resize数组大小
        offset = first.p-zl;
        zl = ziplistResize(zl, intrev32ifbe(ZIPLIST_BYTES(zl))-totlen+nextdiff);
        //修改链表元素个数
        ZIPLIST_INCR_LENGTH(zl,-deleted);
        p = zl+offset;
        //nextdiff != 0表示元素大小发生变化，需要进行级联更新
        if (nextdiff != 0)
            zl = __ziplistCascadeUpdate(zl,p);
    }
    return zl;
}

最后我们看下元素的查找操作：

//参数依次为：压缩链表，要查找元素的值，要查找元素的长度，每次比较之间跳过的元素个数
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;
        //查询链表当前元素的prevlen字段
        ZIP_DECODE_PREVLENSIZE(p, prevlensize);
        //查询链表当前元素的encoding字段
        ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
        q = p + prevlensize + lensize;
        //已到达需要比较的元素位置
        if (skipcnt == 0) {
            //如果链表中的当前元素时字符串
            if (ZIP_IS_STR(encoding)) {
                //跟要查找的字符串进行比较
                if (len == vlen && memcmp(q, vstr, vlen) == 0) {
                    //匹配成功，则要查找元素的指针
                    return p;
                }
            } else {
                //如果当前元素为数字且vencoding为0
                if (vencoding == 0) {
                    //尝试对要查找的值进行数字编码
                    if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {
                        //如果编码失败，则说明要查找的元素根本不是数字
                        //然后把vencoding设置为最大值，起一个标记作用
                        //也就是说后面就不用尝试把要查找的值编码成数字了
                        vencoding = UCHAR_MAX;
                    }
                    assert(vencoding);
                }
                //如果vencoding != UCHAR_MAX，则说明要查找的元素成功编码为数字
                if (vencoding != UCHAR_MAX) {
                    //按数字取出当前链表中的元素
                    long long ll = zipLoadInteger(q, encoding);
                    if (ll == vll) {
                        //如果两个数字相等，则返回元素指针
                        return p;
                    }
                }
            }
            //重置需要跳过的元素个数
            skipcnt = skip;
        } else {
            //跳过元素
            skipcnt--;
        }
        //遍历下个元素
        p = q + len;
    }
    //遍历完整个链表，没有找到元素
    return NULL;
}

到这里就把压缩链表的创建，添加，删除，查找四个基本操作原理总结完了。

三、压缩链表ziplist数据结构总结

压缩链表ziplist在redis中的应用非常广泛，它算是redis中最具特色的数据结构了。该数据结构的精髓其实就在于文章第一部分总结的编码规则，先理清楚这部分内容，后面的源码可以简单看下加深理解。

不得不说源码部分着实有点冗长，确实需要耐心，我自己在读的过程中也很头大。如果对源码感兴趣的话，建议按我的方法，先自己梳理某个操作（例如上面提到的插入元素）需要做哪些事情，然后再看代码可能会更好理解一些。

---

最后留一个小问题，既然redis中的ziplist对内存利用率如此之高，那为什么还要提供普通的链表结构供用户使用呢？
这个问题没有标准答案，仁者见仁智者见智吧~