结构

ziplist 在内存中的排列格式如下：

<zlbytes> <zltail> <zllen> <zlentry>  ... <zlentry> <zlend>

可以看出 ziplist 主要由四种不同的结构组成

  • zlbytes：占用4字节，记录了整个 ziplist 字节数长度；
  • zltail：占用4字节，记录 ziplist 结构初始位置到末尾的 zlentry 的字节数偏移，通过这个值加上压缩列表起始位置的指针可以快速定位到末尾的数据节点；
  • zllen：占用2个字节，记录了压缩列表中数据节点 zlentry 的总数，最大可以记录 2^16 -1 = 655355，当 zllen == 655355 时候，如果要获取列表长度，就需要遍历整个压缩列表的节点；
  • zlentry：压缩列表中的数据节点，节点长度由保存的数据来决定；
  • zlend：占用1个字节，标记压缩列表的末端，固定等于 0xFF（十进制255）

其中数据节点 zlentry 的组成结构可以细分为：

<prevlen> <encoding> <entry-data>

如果当前节点没有保存数据，那结构就是：

<prevlen> <encoding>
  • prevlen：以字节未单位，记录前一个数据节点的长度值，具体：
    • 如果前一节点长度小于254个字节，prevlen 只占用1个字节（1个字节最大可以表示2^8-1=255，而255是固定表示末尾不能采用，所以最大记录到254）；
    • 如果前一个节点长度大于5个字节，则 prevlen 会占用5个字节，其中第1个字节固定为0xFE（十进制254），后面4个字节记录具体长度值。
  • encoding：记录当前节点的entry-data的数据类型和字节长度长度，具体：
    • 前两位分别为00,01和10表示当前节点存放的是字节数组，当前encoding字段分别对应占用1字节、2字节和5字节，除去初始的2bit后剩余bit位记录了entry-data部分的字节数长度；
    • 如果前两位是11，表示当前节点存放的是整数类型，entry-data 占用了1个字节，具体的整数类型由去掉初始 2bit 后剩余的其他 bit 来表示；
  • entry-data：用于具体存放要保存的数据，占用长度不定。

连锁更新

由于 ziplist 中的每个数据节点都存放了前一节点的长度（prevlen），而prevlen的字节长度又是不确定的：

  • 如果前一节点小于254个字节，prevlen长度占用1字节；
  • 如果前一节点大于或者等于254个字节，prevlen 又需要占用5个字节；

如果当 ziplist 中有多个连续的节点 E1 ~ En 长度都处于250~253个字节（此时都只需要用1个字节来存储 prevlen），这时候如果在其中一个节点(如 E1)之前插入某个长度大于254个字节的节点时候，E1 节点的 prevlen 就需要从1个字节扩展到5个字节长度，更新后 E1 总长度为 254~257 之间，就会发现 E1 节点也出现了“超额”的现象，为了满足约定的规则，那么 E2 节点的 prevlen 长度也需要更新……最终 E1~En 之间的所有节点都会触发连锁更新。

这种在增减节点时候导致 ziplist 中连续节点都扩展空间的行为，就是连锁更新。每次连锁更新的节点都需要重新申请分配内存空间，这和 ziplist 节省内存空间，高效利用内存的初衷是相悖的。

但是直到 Redis 7.0 推出 listpack（紧凑列表）来替代 ziplist 之前，ziplist 还是在 Redis 底层承载了很多类型的数据，并没有因为连锁更新的问题被淘汰，主要原因是因为：

  • 连锁更新触发的条件其实是比较苛刻的，需要多个连续的、长度在 205~253 之间的节点；
  • Redis 通常只会在数据量比较小时候使用 ziplist 来存放数据，因此，即使出现了连锁更新现象，少数节点发生连续分配内存行为，事实上对性能层面的影响也是很有限的。

源码分析

先来看看 ziplist 结构体定义，代码在 https://github.com/redis/redis/blob/7.2.4/src/ziplist.h#L38：

// ziplist 每个节点有字符串和整数两种情况
// 需要同时记录两种字段
typedef struct {
    // 如果元素是字符串，sval表示字符串值，slen未字符串长度
    unsigned char *sval;
    unsigned int slen;
    // 如果存放的整数，只使用lval来记录存放的整数值
    long long lval;
} ziplistEntry;

zlentry 数据节点的定义：

typedef struct zlentry {
    /* 记录prelen占用的字节数，可能为1或者5 */
    unsigned int prevrawlensize;
    /* prelen的具体数值，即前一个节点字节数长度 */
    unsigned int prevrawlen;

    /*
    * 记录当前entry保存的数据类型或者长度占用的字节数
    * 保存string数据时候，可能为1,2,5
    * 保存integer数据时候，固定为1
    */
    unsigned int lensize;
    /*
    * 当前zlentry的实际长度
    * string类型的内容，len为字符串长度
    * integer类型的内容，根据数值范围可能是1,2,3,4,8
    */
    unsigned int len;
    /* prevrawlensize + lensize. */
    unsigned int headersize;
    /* ZIP_STR_* 或者 ZIP_INT_* 定义的编码类型 */
    unsigned char encoding;
    /* 指向entry最初始位置的指针 */
    unsigned char *p;
} zlentry;

定义了一些宏来获取 ziplist 的各个重要数据：

/* 获取 zlbytes */
#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl)))

/* 获取zltail值 */
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))

/* 获取zllen值 */
#define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))

/* 获取header部分长度，即 zlbytes+zltail+zllen的总长度，固定是10字节 */
#define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))

/* 获取zllen的长度，固定为1字节 */
#define ZIPLIST_END_SIZE        (sizeof(uint8_t))

/* 获取指向第一个zlentry节点的指针 */
#define ZIPLIST_ENTRY_HEAD(zl)  ((zl)+ZIPLIST_HEADER_SIZE)

/* 获取最后一个zlentry节点的指针，通过zltail计算得出 */
#define ZIPLIST_ENTRY_TAIL(zl)  ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl)))

/* 获取指向ziplist最后一个字节的指针，也即是zlend的末端指针 */
#define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-ZIPLIST_END_SIZE)

再来看看向 ziplist 中插入新节点时候可能触发连锁更新的情况：

/*
* 这个函数用来执行新节点插入操作
*/
unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {
    ...
    /*
    * 这之前进行了一系列计算，或者当前节点更新后，下个节点的的prelen也需要更新，就需要触发连锁更新的逻辑
    */
    if (nextdiff != 0) {
        offset = p-zl;
        zl = __ziplistCascadeUpdate(zl,p+reqlen);
        p = zl+offset;
    }
    ...
}

具体执行连锁更新的代码：

unsigned char *__ziplistCascadeUpdate(unsigned char *zl, unsigned char *p) {
    ...
    // 依次遍历，直到表尾结点
    while (p[0] != ZIP_END) {
        ...
        // 如果要更新的节点足够承载要更新的prevlen长度，则不用扩容或者缩容，更新后停止
        if (cur.prevrawlensize >= prevlensize) {
            if (cur.prevrawlensize == prevlensize) {
                zipStorePrevEntryLength(p, prevlen);
            } else {
                /* This would result in shrinking, which we want to avoid.
                 * So, set "prevlen" in the available bytes. */
                zipStorePrevEntryLengthLarge(p, prevlen);
            }
            break;
        }
        ...
        // 上面的条件不满足，就只能依次更新
        prevlensize = zipStorePrevEntryLength(NULL, prevlen);
        ...
    }
    ...
}

最坏的情况下，连锁更新会导致从表头节点到表尾节点的所有节点都更新一遍，每次更新都要带来依次内存重新分配的过程，对 ziplist 进行增删操作时候最坏的时间复杂度会是 O(N^2)级别，可见压缩列表对于需要频繁修改的数据本身是不太友好的，ziplist 设计的初衷本来就不是用来存储大量的、频繁修改的内容的，其优势在于高效利用内存，防止了内存碎片的问题，一定程度上体现了“时间换空间”的思想。