压缩列表 ZipList

睡不醒的鲤鱼 2021-05-15 Redis Redis 原理

压缩列表「ZipList」是一个经过特殊编码的双向链表，它的设计目标就是为了提高存储效率。

压缩列表可用于存储字符串或整数，其中整数是按真正的二进制表示进行编码的，而不是编码成字符串序列，它能以 $O(1)$ 的时间复杂度在表的两端提供 push 和 pop 操作。

# 一、数据结构

# 1.1 压缩列表

Redis 使用字节数组表示一个压缩列表，压缩列表结构示意图如下所示。

各字段的含义如下：

zlbytes：压缩列表的字节长度，占 4 个字节，因此压缩列表最多有 $2^{32}-1$ 个字节；
zltail：压缩列表尾元素相对于压缩列表起始地址的偏移量，占 4 个字节；
zllen：压缩列表的元素个数，占 2 个字节；
entry：压缩列表存储的元素，可以是字节数组或者整数，长度不限，entry 的编码结构在后面详细介绍；
zlend：压缩列表的结尾，占 1 个字节，恒为 0xFF。

注意

zllen 无法存储元素个数超过 $65535(2^{16}-1)$ 的压缩列表，必须遍历整个压缩列表才能获取到元素个数。

假设 char* zl 指向压缩列表首地址，Redis 可通过以下宏定义实现压缩列表各个字段的存取操作。

/* Return total bytes a ziplist is composed of. */
// zl 指向 zlbytes 字段
#define ZIPLIST_BYTES(zl)       (*((uint32_t*)(zl)))

/* Return the offset of the last item inside the ziplist. */
// zl+4 指向 zltail 字段
#define ZIPLIST_TAIL_OFFSET(zl) (*((uint32_t*)((zl)+sizeof(uint32_t))))

/* Return the length of a ziplist, or UINT16_MAX if the length cannot be
 * determined without scanning the whole ziplist. */
// zl+8 指向 zllen 字段，返回最大值 UINT16_MAX 时，需遍历获取长度
#define ZIPLIST_LENGTH(zl)      (*((uint16_t*)((zl)+sizeof(uint32_t)*2)))

/* The size of a ziplist header: two 32 bit integers for the total
 * bytes count and last item offset. One 16 bit integer for the number
 * of items field. */
// 返回 ziplist 表头大小
#define ZIPLIST_HEADER_SIZE     (sizeof(uint32_t)*2+sizeof(uint16_t))

/* Size of the "end of ziplist" entry. Just one byte. */
// 返回 ziplist 表尾大小
#define ZIPLIST_END_SIZE        (sizeof(uint8_t))

/* Return the pointer to the first entry of a ziplist. */
// 指向第一个节点起始位置
#define ZIPLIST_ENTRY_HEAD(zl)  ((zl)+ZIPLIST_HEADER_SIZE)

/* Return the pointer to the last entry of a ziplist, using the
 * last entry offset inside the ziplist header. */
// 指向尾元素首地址，intrev32ifbe 使得数据存取统一采用小端法
#define ZIPLIST_ENTRY_TAIL(zl)  ((zl)+intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl)))

/* Return the pointer to the last byte of a ziplist, which is, the
 * end of ziplist FF entry. */
// 压缩列表最后一个字节，即为 zlend 字段
#define ZIPLIST_ENTRY_END(zl)   ((zl)+intrev32ifbe(ZIPLIST_BYTES(zl))-1)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

# 1.2 压缩列表节点

上一节介绍了压缩列表的基本结构，我们可以很容易地获得压缩列表的字节长度、元素个数等。

但还有一些重要操作需要借助压缩列表元素 entry 的特殊结构才能实现，如压缩列表的遍历、获取任意存储元素的类型、获取字节数组的长度等，下面来对压缩列表元素的结构详细介绍，其编码结构示意图如下所示：

prevlen 字段

prevlen 字段表示前一个元素的字节长度，占 1 个或 5 个字节，具体规则如下：

前置节点长度小于 254 字节，占用 1 个字节记录该长度值；
前置节点长度超过 254 字节，占用 5 个字节记录该长度值；
- 第 1 个字节的值将被设为 254 ，用于标识这是一个 5 字节长的长度值；
- 之后的 4 个字节则用于保存前置节点的实际长度。

说明：由于 255 被 zlend 占用，用于标识压缩列表结尾，因此这里用 254 来标识。

encoding 字段

encoding 字段表示当前元素的编码，即 entry-data 字段存储的数据类型（字节数组或整数），数据内容存储在 entry-data 字段。为了节约内存，encoding 字段同样长度可变，具体编码形式如下所示：

编码	编码长度	entry-data 属性保存的值
$00$ xxxxxx	1 字节	最大长度为 $2^6-1$ 的字节数组
$01$ xxxxxx xxxxxxxx	2 字节	最大长度为 $2^{14}-1$ 的字节数组
$10$ $000000$ xxxxxxxx xxxxxxxx xxxxxxxx xxxxxxxx	5 字节	最大长度为 $2^{32}-1$ 的字节数组
$11$ $00$ $0000$	1 字节	int16_t 类型整数
$11$ $01$ $0000$	1 字节	int32_t 类型整数
$11$ $10$ $0000$	1 字节	int64_t 类型整数
$11$ $11$ $0000$	1 字节	24 位有符号整数
$11$ $11$ $1110$	1 字节	8 位有符号整数
$11$ $11$ xxxx	1 字节	无 entry-data 字段；xxxx 表示 0~12 的整数

此处可参考以下源码注释帮助对 encoding 定义的理解。

点击查看源码注释

/*
 * The encoding field of the entry depends on the content of the
 * entry. When the entry is a string, the first 2 bits of the encoding first
 * byte will hold the type of encoding used to store the length of the string,
 * followed by the actual length of the string. When the entry is an integer
 * the first 2 bits are both set to 1. The following 2 bits are used to specify
 * what kind of integer will be stored after this header. An overview of the
 * different types and encodings is as follows. The first byte is always enough
 * to determine the kind of entry.
 *
 * |00pppppp| - 1 byte
 *      String value with length less than or equal to 63 bytes (6 bits).
 *      "pppppp" represents the unsigned 6 bit length.
 * |01pppppp|qqqqqqqq| - 2 bytes
 *      String value with length less than or equal to 16383 bytes (14 bits).
 *      IMPORTANT: The 14 bit number is stored in big endian.
 * |10000000|qqqqqqqq|rrrrrrrr|ssssssss|tttttttt| - 5 bytes
 *      String value with length greater than or equal to 16384 bytes.
 *      Only the 4 bytes following the first byte represents the length
 *      up to 32^2-1. The 6 lower bits of the first byte are not used and
 *      are set to zero.
 *      IMPORTANT: The 32 bit number is stored in big endian.
 * |11000000| - 3 bytes
 *      Integer encoded as int16_t (2 bytes).
 * |11010000| - 5 bytes
 *      Integer encoded as int32_t (4 bytes).
 * |11100000| - 9 bytes
 *      Integer encoded as int64_t (8 bytes).
 * |11110000| - 4 bytes
 *      Integer encoded as 24 bit signed (3 bytes).
 * |11111110| - 2 bytes
 *      Integer encoded as 8 bit signed (1 byte).
 * |1111xxxx| - (with xxxx between 0000 and 1101) immediate 4 bit integer.
 *      Unsigned integer from 0 to 12. The encoded value is actually from
 *      1 to 13 because 0000 and 1111 can not be used, so 1 should be
 *      subtracted from the encoded 4 bit value to obtain the right value.
 * |11111111| - End of ziplist special entry.
 *
 * Like for the ziplist header, all the integers are represented in little
 * endian byte order, even when this code is compiled in big endian systems.
 */

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41

可以看出，根据 encoding 字段第 1 个字节的前 2 位，可以判断 entry-data 字段存储的是整数还是字节数组。

当 entry-data 存储的是字节数组时，后续字节标识字节数组的实际长度；
当 entry-data 存储的是整数时，可根据第 3、4 位判断整数的具体类型；
当 encoding 字段标识当前元素存储的是 0～12 的立即数时，数据直接存储在 encoding 字段的最后 4 位，此时没有 entry-data 字段。

注意

为什么是 0～12 ？

xxxx 的值只能在 $0001$ 和 $1101$ 之间，即从 1 到 13 一共 13 个值，否则会与其他编码情况冲突，而小数值应该从 0 开始，因此这 13 个值分别表示 0 到 12，即 xxxx 的值减去 1 才是它所要表示的那个整数数据的值。

参照 encoding 字段的编码表格，Redis 预定义了以下常量对应 encoding 字段的各编码类型：

#define ZIP_STR_06B (0 << 6)
#define ZIP_STR_14B (1 << 6)
#define ZIP_STR_32B (2 << 6)
#define ZIP_INT_16B (0xc0 | 0<<4)
#define ZIP_INT_32B (0xc0 | 1<<4)
#define ZIP_INT_64B (0xc0 | 2<<4)
#define ZIP_INT_24B (0xc0 | 3<<4)
#define ZIP_INT_8B 0xfe

1
2
3
4
5
6
7
8

上面介绍了压缩列表的存储结构，可以看到，对于压缩列表的任意元素，获取前一个元素的长度、判断存储的数据类型、获取数据内容都需要复杂的解码运算。解码后的结果应该被缓存起来，为此定义了结构体 zlentry，用于表示解码后的压缩列表元素。

/* We use this function to receive information about a ziplist entry.
 * Note that this is not how the data is actually encoded, is just what we
 * get filled by a function in order to operate more easily. */
typedef struct zlentry {
    // 编码 prevrawlen 所需的字节大小
    unsigned int prevrawlensize; /* Bytes used to encode the previous entry len*/
    // 前置节点的长度
    unsigned int prevrawlen;     /* Previous entry len. */
    // 编码 len 所需的字节大小
    unsigned int lensize;        /* Bytes used to encode this entry type/len.
                                    For example strings have a 1, 2 or 5 bytes
                                    header. Integers always use a single byte.*/
    // 当前节点的长度
    unsigned int len;            /* Bytes used to represent the actual entry.
                                    For strings this is just the string length
                                    while for integers it is 1, 2, 3, 4, 8 or
                                    0 (for 4 bit immediate) depending on the
                                    number range. */
    // 当前节点 header 大小
    unsigned int headersize;     /* prevrawlensize + lensize. */
    // 当前节点值所使用的编码类型
    unsigned char encoding;      /* Set to ZIP_STR_* or ZIP_INT_* depending on
                                    the entry encoding. However for 4 bits
                                    immediate integers this can assume a range
                                    of values and must be range-checked. */
    // 指向当前节点的指针
    unsigned char *p;            /* Pointer to the very start of the entry, that
                                    is, this points to prev-entry-len field. */
} zlentry;

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29

函数 zipEntry 用来解码压缩列表的元素，存储于 zlentry 结构体 e。

/* Return a struct with all information about an entry. */
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;
}

1
2
3
4
5
6
7
8

可以看到，解码主要可以分为以下两个步骤：

解码 prevlen，此时入参 ptr 指向元素首地址。

点击查看代码

#define ZIP_BIG_PREVLEN 254 /* Max number of bytes of the previous entry, for
                               the "prevlen" field prefixing each entry, to be
                               represented with just a single byte. Otherwise
                               it is represented as FF AA BB CC DD, where
                               AA BB CC DD are a 4 bytes unsigned integer
                               representing the previous entry len. */

/* Return the number of bytes used to encode the length of the previous
 * entry. The length is returned by setting the var 'prevlensize'. */
#define ZIP_DECODE_PREVLENSIZE(ptr, prevlensize) do {                          \
    if ((ptr)[0] < ZIP_BIG_PREVLEN) {                                          \
        (prevlensize) = 1;                                                     \
    } else {                                                                   \
        (prevlensize) = 5;                                                     \
    }                                                                          \
} while(0);

/* Return the length of the previous element, and the number of bytes that
 * are used in order to encode the previous element length.
 * 'ptr' must point to the prevlen prefix of an entry (that encodes the
 * length of the previous entry in order to navigate the elements backward).
 * The length of the previous entry is stored in 'prevlen', the number of
 * bytes needed to encode the previous entry length are stored in
 * 'prevlensize'. */
#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);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

解码 encoding 字段，此时入参 ptr 指向元素首地址偏移 prevlen 长度的位置。

点击查看代码

#define ZIP_STR_MASK 0xc0

/* Extract the encoding from the byte pointed by 'ptr' and set it into
 * 'encoding' field of the zlentry structure. */
#define ZIP_ENTRY_ENCODING(ptr, encoding) do {  \
    (encoding) = (ptr[0]); \
    if ((encoding) < ZIP_STR_MASK) (encoding) &= ZIP_STR_MASK; \
} while(0)

/* Return bytes needed to store integer encoded by 'encoding'. */
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;
}

/* Decode the entry encoding type and data length (string length for strings,
 * number of bytes used for the integer for integer entries) encoded in 'ptr'.
 * The 'encoding' variable will hold the entry encoding, the 'lensize'
 * variable will hold the number of bytes required to encode the entry
 * length, and the 'len' variable will hold the entry length. */
#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);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# 二、基本操作

# 2.1 创建压缩列表

创建空的压缩列表，只需要分配初始存储空间 ZIPLIST_HEADER_SIZE+1 个字节，并对 zlbytes、zltail、zllen 和 zlend 字段初始化即可。

/* Create a new empty ziplist. */
unsigned char *ziplistNew(void) {
    
    // ZIPLIST_HEADER_SIZE 是 ziplist 表头的大小
    // 1 字节是表末端 zlend 的大小
    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
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

# 2.2 插入元素

压缩列表插入元素的 API 定义如下，函数输入参数 zl 表示压缩列表首地址，p 指向元素插入位置，s 表示数据内容，slen 表示数据长度，返回值为压缩列表首地址。

/**
 * 将包含给定值 s 的新节点插入到给定的位置 p 中；
 */
unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p,
                             unsigned char *s, unsigned int slen) {
    return __ziplistInsert(zl,p,s,slen);
}

1
2
3
4
5
6
7

插入元素可以简要分为 3 个步骤：编码、重新分配空间、数据复制，下面分别讲解每个步骤的实现逻辑。

1. 编码

编码即计算 prevlen、encoding 和 entry-data 的内容。

其中 prevlen 的计算需要根据插入位置分情况讨论，插入位置示意图如下所示：

当压缩列表为空、插入位置为 $P_0$ 时，不存在前一个元素，即前一个元素的长度为 0；
当插入位置为 $P_1$ 时，需要获取 $x_i$ 元素的长度，而 $x_{i+1}$ 元素的 prevlen 字段存储的就是 $x_i$ 元素的长度，很容易获取到；
当插入位置为 $P_2$ 时，需要获取 $x_n$ 元素的长度， $x_n$ 是压缩列表的尾元素，计算元素长度时需要将其 3个字段长度相加，函数实现如下。

/* Return the total number of bytes used by the entry pointed to by 'p'. */
unsigned int zipRawEntryLength(unsigned char *p) {
    unsigned int prevlensize, encoding, lensize, len;
    ZIP_DECODE_PREVLENSIZE(p, prevlensize);
    ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
    return prevlensize + lensize + len;
}

1
2
3
4
5
6
7

encoding 字段标识的是当前元素存储的数据类型和数据长度。编码时首先尝试将数据内容解析为整数，如果解析成功，则按照压缩列表整数类型编码存储；如果解析失败，则按照压缩列表字节数组类型编码存储。

    /* 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 zipStoreEntryEncoding will use the
         * string length to figure out how to encode it. */
        reqlen = slen;
    }
    /* We need space for both the length of the previous entry and
     * the length of the payload. */
    reqlen += zipStorePrevEntryLength(NULL,prevlen);
    reqlen += zipStoreEntryEncoding(NULL,encoding,slen);

1
2
3
4
5
6
7
8
9
10
11
12
13

上述程序尝试按照整数解析新添加元素的数据内容，数值存储在变量 value 中，编码存储在变量 encoding 中。如果解析成功，还需要计算整数所占字节数。

变量 reqlen 最终存储的是当前元素所需空间大小，初始赋值为元素 content 字段所需空间大小，再累加 prevlen 和 encoding 字段所需空间大小。

2. 重新分配空间

由于新插入了元素，压缩列表所需空间增大，因此需要重新分配存储空间，一种压缩列表长度变化情况如下所示：

插入元素前， $x_i$ 元素的长度为 128 字节， $x_{i+1}$ 元素的 prevlen 字段占 1 个字节；添加元素 $x_{new}$ ，元素长度为 1024 字节，此时 $x_{i+1}$ 元素的 prevlen 字段需要占 5 个字节，即压缩列表的长度不仅增加了 1024 个字节，还要加上 $x_{i+1}$ 元素扩展的 4 个字节。

由于重新分配了空间，新元素插入的位置指针 P 会失效，可以预先计算好指针 P 相对于压缩列表首地址的偏移量，待分配空间之后再偏移即可。

3. 数据复制

重新分配空间之后，需要将位置 P 后的元素移动到指定位置，将新元素插入到位置 P。

# 2.3 删除元素

压缩列表删除元素的 API 定义如下，函数输入参数 zl 指向压缩列表首地址，*p 指向待删除元素的首地址（参数 p 同时可以作为输出参数），返回参数为压缩列表首地址。

/* Delete a single entry from the ziplist, pointed to by *p.
 * Also update *p in place, to be able to iterate over the
 * ziplist, while deleting entries. */
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;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14

删除元素同样可以简要分为 3 个步骤：计算待删除元素的总长度、数据复制、重新分配空间，下面分别讲解每个步骤的实现逻辑。

1. 计算待删除元素的总长度

    // 解码第一个待删除元素
    zipEntry(p, &first);

    // 遍历所有待删除元素，同时指针 p 向后偏移
    for (i = 0; p[0] != ZIP_END && i < num; i++) {
        p += zipRawEntryLength(p);
        deleted++;
    }

    // totlen 即为待删除元素总长度
    totlen = p-first.p; /* Bytes taken by the element(s) to delete. */

1
2
3
4
5
6
7
8
9
10
11

2. 数据复制

第 1 步完成之后，指针 first 与指针 p 之间的元素都是待删除的。当指针 p 恰好指向 zlend 字段时，不再需要复制数据，只需要更新尾节点的偏移量即可。

下面分析另一种情况，即指针 p 指向的是某一个元素，而不是 zlend 字段，一种压缩列表长度变化情况如下所示：

删除元素 $x_{i+1}$ 到元素 $x_{n-1}$ 之间的 n-x-1 个元素，元素 $x_{n-1}$ 的长度为 12 字节，因此 $x_n$ 的 prevlen 字段占 1 字节；删除这些元素之后， $x_i$ 成为 $x_n$ 的前一个元素，元素 $x_i$ 的长度为 512 字节，因此 $x_n$ 的 prevlen 字段需要占 5 个字节，即删除元素之后的压缩列表总长度还与元素 $x_n$ 的长度变化量有关。

3. 重新分配空间

与插入元素类似，这里不再赘述。

# 2.4 遍历压缩列表

压缩列表的遍历 API 定义如下，函数输入参数 zl 指向压缩列表首地址，p 指向当前访问元素的首地址；ziplistNext 函数返回后一个元素的首地址，ziplistPrev 返回前一个元素的首地址。

// 后向遍历
unsigned char *ziplistNext(unsigned char *zl, unsigned char *p);

// 前向遍历
unsigned char *ziplistPrev(unsigned char *zl, unsigned char *p);

1
2
3
4
5

压缩列表每个元素的 prevlen 字段存储的是前一个元素的长度，因此压缩列表的前向遍历相对简单，表达式 p-prevlen 即可获取前一个元素的首地址，这里不做详述。

后向遍历时，需要解码当前元素，计算当前元素的长度，才能获取后一个元素首地址，ziplistNext 函数实现如下：

/* Return pointer to next entry in ziplist.
 *
 * zl is the pointer to the ziplist
 * p is the pointer to the current element
 *
 * The element after 'p' is returned, otherwise NULL if we are at the end. */
unsigned char *ziplistNext(unsigned char *zl, unsigned char *p) {
    ((void) zl);

    /* "p" could be equal to ZIP_END, caused by ziplistDelete,
     * and we should return NULL. Otherwise, we should return NULL
     * when the *next* element is ZIP_END (there is no next entry). */
    if (p[0] == ZIP_END) {
        return NULL;
    }

    p += zipRawEntryLength(p);
    if (p[0] == ZIP_END) {
        return NULL;
    }

    return p;
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

# 三、连锁更新

压缩列表连锁更新示意如下所示，当删除压缩列表 zl1 位置 $p_1$ 的元素 $x_i$ ，或者在压缩列表 zl2 位置 $p_2$ 插入元素 $x_j$ 时，会导致连锁更新。

压缩列表 zl1 中，元素 $x_i$ 之后的元素 $x_{i+1}$ 、 $x_{i+2}$ 的长度都是 253 字节，因此这些元素的 prevlen 字段长度都是 1 字节。

当删除元素 $x_i$ 时，元素 $x_{i+1}$ 的前驱节点改为元素 $x_{i-1}$ ，长度为 512 字节，元素 $x_{i+1}$ 的 prevlen 字段需要 5 字节才能存储元素 $x_{i-1}$ 的长度，则元素 $x_{i+1}$ 的长度需要扩展至 257 字节；而由于元素 $x_{i+1}$ 长度的增大，元素 $x_{i+2}$ 的 prevlen 字段同样需要改变。

依此类推，由于删除了元素 $x_i$ ，之后的元素 $x_{i+1}$ 、 $x_{i+2}$ 的长度都必须扩展，而每次扩展都将重新分配内存，导致效率很低。

压缩列表 zl2 中，插入元素 $x_j$ 时同样会出现这种情况，称为连锁更新。

注意

连锁更新造成性能问题的几率是很低的。

首先，压缩列表里要恰好有多个连续的、长度介于 250 字节至 253 字节之间的节点，连锁更新才可能触发，在实际中，这种情况并不多见；
其次，即使出现连锁更新，但只要被更新的节点不多，就不会对性能造成任何影响。

# 四、总结

压缩列表是一种为节约内存而开发的顺序型数据结构；
压缩列表可以包含多个节点，每个节点可以保存一个字节数组或整数值；
添加或删除节点到压缩列表，可能导致连锁更新，但这种操作出现几率并不高。

# 五、参考资料

《Redis 5 设计与源码分析》– 陈雷
《Redis 设计与实现》– 黄健宏
Redis内部数据结构详解(4) – ziplist (opens new window)

跳跃表 SkipList 哈希表 Dict

鲤鱼笔记

Choose mode