Redis 核心原理与实践 -- 列表实现原理之 ziplist 结构

Redis 列表类型可以存储一组按插入顺序排序的字符串，它非常灵活，支持在两端插入、弹出数据，可以充当栈和队列的角色。

> LPUSH fruit apple(integer) 1> RPUSH fruit banana(integer) 2> RPOP fruit"banana"> LPOP fruit"apple"

本文探讨 Redis 中列表类型的实现。

ziplist

使用数组和链表结构都可以实现列表类型。Redis 中使用的是链表结构。下面是一种常见的链表实现方式 adlist.h：

typedef struct listNode {    struct listNode *prev;    struct listNode *next;    void *value;} listNode;
typedef struct list {    listNode *head;    listNode *tail;    void *(*dup)(void *ptr);    void (*free)(void *ptr);    int (*match)(void *ptr, void *key);    unsigned long len;} list;

Redis 内部使用该链表保存运行数据，如主服务下所有的从服务器信息。

但 Redis 并不使用该链表保存用户列表数据，因为它对内存管理不够友好：

（1）链表中每一个节点都占用独立的一块内存，导致内存碎片过多。

（2）链表节点中前后节点指针占用过多的额外内存。

读者可以思考一下，用什么结构可以比较好地解决上面的两个问题？没错，数组。ziplist 是一种类似数组的紧凑型链表格式。它会申请一整块内存，在这个内存上存放该链表所有数据，这就是 ziplist 的设计思想。

定义

ziplist 总体布局如下：

<zlbytes> <zltail> <zllen> <entry> <entry> ... <entry> <zlend>

• zlbytes：uint32_t，记录整个 ziplist 占用的字节数，包括 zlbytes 占用的 4 字节。

• zltail：uint32_t，记录从 ziplist 起始位置到最后一个节点的偏移量，用于支持链表从尾部弹出或反向（从尾到头）遍历链表。

• zllen：uint16_t，记录节点数量，如果存在超过 216-2 个节点，则这个值设置为 216-1，这时需要遍历整个 ziplist 获取真正的节点数量。

• zlend：uint8_t，一个特殊的标志节点，等于 255，标志 ziplist 结尾。其他节点数据不会以 255 开头。

entry 就是 ziplist 中保存的节点。entry 的格式如下：

<prevlen> <encoding> <entry-data>

• entry-data：该节点元素，即节点存储的数据。

• prevlen：记录前驱节点长度，单位为字节，该属性长度为 1 字节或 5 字节。

• ① 如果前驱节点长度小于 254，则使用 1 字节存储前驱节点长度。

• ② 否则，使用 5 字节，并且第一个字节固定为 254，剩下 4 个字节存储前驱节点长度。

• encoding：代表当前节点元素的编码格式，包含编码类型和节点长度。一个 ziplist 中，不同节点元素的编码格式可以不同。编码格式规范如下：

• ① 00pppppp（pppppp 代表 encoding 的低 6 位，下同）：字符串编码，长度小于或等于 63（26-1），长度存放在 encoding 的低 6 位中。

• ② 01pppppp：字符串编码， 长度小于或等于 16383（214-1），长度存放在 encoding 的后 6 位和 encoding 后 1 字节中。

• ③ 10000000：字符串编码，长度大于 16383（214-1），长度存放在 encoding 后 4 字节中。

• ④ 11000000：数值编码， 类型为 int16_t，占用 2 字节。

• ⑤ 11010000：数值编码，类型为 int32_t，占用 4 字节。

• ⑥ 11100000：数值编码，类型为 int64_t，占用 8 字节。

• ⑦ 11110000：数值编码，使用 3 字节保存一个整数。

• ⑧ 11111110：数值编码，使用 1 字节保存一个整数。

• ⑨ 1111xxxx：使用 encoding 低 4 位存储一个整数，存储数值范围为 0～12。该编码下 encoding 低 4 位的可用范围为 0001～1101，encoding 低 4 位减 1 为实际存储的值。

• ⑩ 11111111：255，ziplist 结束节点。

• 注意第②、③种编码格式，除了 encoding 属性，还需要额外的空间存储节点元素长度。第⑨种格式也比较特殊，节点元素直接存放在 encoding 属性上。该编码是针对小数字的优化。这时 entry-data 为空。

字节序

encoding 属性使用多个字节存储节点元素长度，这种多字节数据存储在计算机内存中或者进行网络传输时的字节顺序称为字节序，字节序有两种类型：大端字节序和小端字节序。

• 大端字节序：低字节数据保存在内存高地址位置，高字节数据保存在内存低地址位置。

• 小端字节序：低字节数据保存在内存低地址位置，高字节数据保存在内存高地址位置。

数值 0X44332211 的大端字节序和小端字节序存储方式如图 2-1 所示。

CPU 处理指令通常是按照内存地址增长方向执行的。使用小端字节序，CPU 可以先读取并处理低位字节，执行计算的借位、进位操作时效率更高。大端字节序则更符合人们的读写习惯。

ziplist 采取的是小端字节序。

下面是 Redis 提供的一个简单例子：

• [0f 00 00 00]：zlbytes 为 15，代表整个 ziplist 占用 15 字节，注意该数值以小端字节序存储。

• [0c 00 00 00]：zltail 为 12，代表从 ziplist 起始位置到最后一个节点（[02 f6]）的偏移量。

• [02 00]：zllen 为 2，代表 ziplist 中有 2 个节点。

• [00 f3]：00 代表前一个节点长度，f3 使用了 encoding 第⑨种编码格式，存储数据为 encoding 低 4 位减 1，即 2。

• [02 f6]：02 代表前一个节点长度为 2 字节，f5 编码格式同上，存储数据为 5。

• [ff]：结束标志节点。ziplist 是 Redis 中比较复杂的数据结构，希望读者结合上述属性说明和例子，理解 ziplist 中数据的存放格式。

操作分析

提示：本节以下代码如无特殊说明，均在 ziplist.h、ziplist.c 中。

ziplistFind 函数负责在 ziplist 中查找元素：

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;        // [1]        ZIP_DECODE_PREVLENSIZE(p, prevlensize);        // [2]        ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);        q = p + prevlensize + lensize;
        if (skipcnt == 0) {            // [3]            if (ZIP_IS_STR(encoding)) {                if (len == vlen && memcmp(q, vstr, vlen) == 0) {                    return p;                }            } else {                // [4]                if (vencoding == 0) {                    if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {                        vencoding = UCHAR_MAX;                    }                    assert(vencoding);                }
                // [5]                if (vencoding != UCHAR_MAX) {                    long long ll = zipLoadInteger(q, encoding);                    if (ll == vll) {                        return p;                    }                }            }
            // [6]            skipcnt = skip;        } else {            skipcnt--;        }
        // [7]        p = q + len;    }
    return NULL;}

参数说明：

• p：指定从 ziplist 哪个节点开始查找。

• vstr、vlen：待查找元素的内容和长度。

• skip：间隔多少个节点才执行一次元素对比操作。

【1】计算当前节点 prevlen 属性长度是 1 字节还是 5 字节，结果存放在 prevlensize 变量中。

【2】计算当前节点相关属性，结果存放在如下变量中：

encoding：节点编码格式。

lensize：额外存放节点元素长度的字节数，第②、③种格式的 encoding 编码需要额外的空间存放节点元素长度。

len：节点元素的长度。

【3】如果当前节点元素是字符串编码，则对比 String 的内容，若相等则返回。

【4】当前节点元素是数值编码，并且还没有对待查找内容 vstr 进行编码，则对它进行编码操作（编码操作只执行一次），编码后的数值存储在 vll 变量中。

【5】如果上一步编码成功（待查找内容也是数值），则对比编码后的结果，否则不需要对比编码结果。zipLoadInteger 函数从节点元素中提取节点存储的数值，与上一步得到的 vll 变量进行对比。

【6】skipcnt 不为 0，直接跳过节点并将 skipcnt 减 1，直到 skipcnt 为 0 才对比数据。

【7】p 指向 p + prevlensize + lensize + len（数据长度），得到下一个节点的起始位置。

提示：由于源码中部分函数太长，为了版面整洁，本书将其划分为多个代码段，并使用“// more”标志该函数后续还有其他代码段，请读者留意该标志。

下面看一下如何在 ziplist 中插入节点：

unsigned char *__ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char *s, unsigned int slen) {    ...    // [1]    if (p[0] != ZIP_END) {        ZIP_DECODE_PREVLEN(p, prevlensize, prevlen);    } else {        unsigned char *ptail = ZIPLIST_ENTRY_TAIL(zl);        if (ptail[0] != ZIP_END) {            prevlen = zipRawEntryLength(ptail);        }    }
    // [2]    if (zipTryEncoding(s,slen,&value,&encoding)) {        reqlen = zipIntSize(encoding);    } else {        reqlen = slen;    }
    // [3]    reqlen += zipStorePrevEntryLength(NULL,prevlen);    reqlen += zipStoreEntryEncoding(NULL,encoding,slen);
    // [4]    int forcelarge = 0;    nextdiff = (p[0] != ZIP_END) ? zipPrevLenByteDiff(p,reqlen) : 0;    if (nextdiff == -4 && reqlen < 4) {        nextdiff = 0;        forcelarge = 1;    }
    // more}

参数说明：

• zl：待插入 ziplist。

• p：指向插入位置的后驱节点。

• s、slen：待插入元素的内容和长度。

【1】计算前驱节点长度并存放到 prevlen 变量中。

如果 p 没有指向 ZIP_END，则可以直接取 p 节点的 prevlen 属性，否则需要通过 ziplist.zltail 找到前驱节点，再获取前驱节点的长度。

【2】对待插入元素的内容进行编码，并将内容的长度存放在 reqlen 变量中。

zipTryEncoding 函数尝试将元素内容编码为数值，如果元素内容能编码为数值，则该函数返回 1，这时 value 指向编码后的值，encoding 存储对应编码格式，否则返回 0。

【3】zipStorePrevEntryLength 函数计算 prevlen 属性的长度（1 字节或 5 字节）。

zipStoreEntryEncoding 函数计算额外存放节点元素长度所需字节数（encoding 编码中第②、③种格式）。reqlen 变量值添加这两个函数的返回值后成为插入节点长度。

【4】zipPrevLenByteDiff 函数计算后驱节点 prevlen 属性长度需调整多少个字节，结果存放在 nextdiff 变量中。

假如 p 指向节点为 e2，而插入前 e2 的前驱节点为 e1，e2 的 prevlen 存储 e1 的长度。

插入后 e2 的前驱节点为插入节点，这时 e2 的 prevlen 应该存储插入节点长度，所以 e2 的 prevlen 需要修改。图 2-2 展示了一个简单示例。

从图 2-2 可以看到，后驱节点 e2 的 prevlen 属性长度从 1 变成了 5，则 nextdiff 变量为 4。如果插入节点长度小于 4，并且原后驱节点 e2 的 prevlen 属性长度为 5，则这时设置 forcelarge 为 1，代表强制保持后驱节点 e2 的 prevlen 属性长度不变。读者可以思考一下，为什么要这样设计？继续分析__ziplistInsert 函数：

    // [5]    offset = p-zl;    zl = ziplistResize(zl,curlen+reqlen+nextdiff);    p = zl+offset;
    if (p[0] != ZIP_END) {        // [6]        memmove(p+reqlen,p-nextdiff,curlen-offset-1+nextdiff);
        // [7]        if (forcelarge)            zipStorePrevEntryLengthLarge(p+reqlen,reqlen);        else            zipStorePrevEntryLength(p+reqlen,reqlen);
        // [8]        ZIPLIST_TAIL_OFFSET(zl) =            intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);        // [9]        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 {        // [10]        ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(p-zl);    }
    // [11]    if (nextdiff != 0) {        offset = p-zl;        zl = __ziplistCascadeUpdate(zl,p+reqlen);        p = zl+offset;    }
    // [12]    p += zipStorePrevEntryLength(p,prevlen);    p += zipStoreEntryEncoding(p,encoding,slen);    if (ZIP_IS_STR(encoding)) {        memcpy(p,s,slen);    } else {        zipSaveInteger(p,value,encoding);    }    // [13]    ZIPLIST_INCR_LENGTH(zl,1);    return zl;}

【5】重新为 ziplist 分配内存，主要是为插入节点申请空间。新 ziplist 的内存大小为 curlen+reqlen+nextdiff（curlen 变量为插入前 ziplist 长度）。将 p 重新赋值为 zl+offset（offset 变量为插入节点的偏移量），是因为 ziplistResize 函数可能会为 ziplist 申请新的内存地址。

下面针对存在后驱节点的场景进行处理。

【6】将插入位置后面所有的节点后移，为插入节点腾出空间。移动空间的起始地址为 p-nextdiff，减去 nextdiff 是因为后驱节点的 prevlen 属性需要调整 nextdiff 长度。移动空间的长度为 curlen-offset-1+nextdiff，减 1 是因为最后的结束标志节点已经在 ziplistResize 函数中设置了。

memmove 是 C 语言提供的内存移动函数。

【7】修改后驱节点的 prevlen 属性。

【8】更新 ziplist.zltail，将其加上 reqlen 的值。

【9】如果存在多个后驱节点，则 ziplist.zltail 还要加上 nextdiff 的值。

如果只有一个后驱节点，则不需要加上 nextdiff，因为这时后驱节点大小变化了 nextdiff，但后驱节点只移动了 reqlen。

提示：zipEntry 函数会将给定节点的所有信息赋值到 zlentry 结构体中。zlentry 结构体用于在计算过程中存放节点信息，实际存储数据格式并不使用该结构体。读者不要被 tail 这个变量名误导，它只是指向插入节点的后驱节点，并不一定指向尾节点。

【10】这里针对不存在后驱节点的场景进行处理，只需更新最后一个节点偏移量 ziplist.zltail。

【11】级联更新。

【12】写入插入数据。

【13】更新 ziplist 节点数量 ziplist.zllen。

解释一下以下代码：

ZIPLIST_TAIL_OFFSET(zl) = intrev32ifbe(intrev32ifbe(ZIPLIST_TAIL_OFFSET(zl))+reqlen);

intrev32ifbe 函数完成以下工作：如果主机使用的小端字节序，则不做处理。如果主机使用的大端字节序，则反转数据字节序（数据第 1 位与第 4 位、第 2 位与第 3 位交换），这样会将大端字节序数据转化为小端字节序，或者将小端字节序数据转化为大端字节序。在上面的代码中，如果主机 CPU 使用的是小端字节序，则 intrev32ifbe 函数不做任何处理。

如果主机 CPU 使用的是大端字节序，则从内存取出数据后，先调用 intrev32ifbe 函数将数据转化为大端字节序后再计算。计算完成后，调用 intrev32ifbe 函数将数据转化为小端字节序后再存入内存。

级联更新

例 2-1：考虑一种极端场景，在 ziplist 的 e2 节点前插入一个新的节点 ne，元素数据长度为 254，如图 2-3 所示。

插入节点如图 2-4 所示。

插入节点后 e2 的 prevlen 属性长度需要更新为 5 字节。

注意 e3 的 prevlen，插入前 e2 的长度为 253，所以 e3 的 prevlen 属性长度为 1 字节，插入新节点后，e2 的长度为 257，那么 e3 的 prevlen 属性长度也要更新了，这就是级联更新。在极端情况下，e3 后续的节点也要继续更新 prevlen 属性。

我们看一下级联更新的实现：

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;    // [1]    while (p[0] != ZIP_END) {        // [2]        zipEntry(p, &cur);        rawlen = cur.headersize + cur.len;        rawlensize = zipStorePrevEntryLength(NULL,rawlen);                      if (p[rawlen] == ZIP_END) break;        // [3]        zipEntry(p+rawlen, &next);                if (next.prevrawlen == rawlen) break;        // [4]        if (next.prevrawlensize < rawlensize) {            // [5]            offset = p-zl;            extra = rawlensize-next.prevrawlensize;            zl = ziplistResize(zl,curlen+extra);            p = zl+offset;
            // [6]            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);            }
            // [7]            memmove(np+rawlensize,                np+next.prevrawlensize,                curlen-noffset-next.prevrawlensize-1);            zipStorePrevEntryLength(np,rawlen);
            // [8]            p += rawlen;            curlen += extra;        } else {            // [9]            if (next.prevrawlensize > rawlensize) {                zipStorePrevEntryLengthLarge(p+rawlen,rawlen);            } else {                // [10]                zipStorePrevEntryLength(p+rawlen,rawlen);            }            // [11]            break;        }    }    return zl;}

参数说明：

• p：p 指向插入节点的后驱节点，为了描述方便，下面将 p 指向的节点称为当前节点。

【1】如果遇到 ZIP_END，则退出循环。

【2】如果下一个节点是 ZIP_END，则退出。

rawlen 变量为当前节点长度，rawlensize 变量为当前节点长度占用的字节数。

p[rawlen]即 p 的后驱节点的第一个字节。

【3】计算后驱节点信息。如果后驱节点的 prevlen 等于当前节点的长度，则退出。

【4】假设存储当前节点长度需要使用 actprevlen（1 或者 5）个字节，这里需要处理 3 种情况。情况 1：后驱节点的 prevlen 属性长度小于 actprevlen，这时需要扩容，如例 2-1 中的场景。

【5】重新为 ziplist 分配内存。

【6】如果后驱节点非 ZIP_END，则需要修改 ziplist.zltail 属性。

【7】将当前节点后面所有的节点后移，腾出空间用来修改后驱节点的 prevlen。

【8】将 p 指针指向后驱节点，继续处理后面节点的 prevlen。

【9】情况 2：后驱节点的 prevlen 属性长度大于 actprevlen，这时需要缩容。为了不让级联更新继续下去，这时强制后驱节点的 prevlen 保持不变。

【10】情况 3：后驱节点的 prevlen 属性长度等于 actprevlen，只要修改后驱节点 prevlen 值，不需要调整 ziplist 的大小。

【11】情况 2 和情况 3 中级联更新不需要继续，退出。

回到上面__ziplistInsert 函数中为什么要设置 forcelarge 为 1 的问题，这样是为了避免插入小节点时，导致级联更新现象的出现，所以强制保持后驱节点的 prevlen 属性长度不变。

从上面的分析我们可以看到，级联更新下的性能是非常糟糕的，而且代码复杂度也高，那么怎么解决这个问题呢？我们先看一下为什么需要使用 prevlen 这个属性？这是因为反向遍历时，每向前跨过一个节点，都必须知道前面这个节点的长度。

既然这样，我们把每个节点长度都保存一份到节点的最后位置，反向遍历时，直接从前一个节点的最后位置获取前一个节点的长度不就可以了吗？而且这样每个节点都是独立的，插入或删除节点都不会有级联更新的现象。基于这种设计，Redis 作者设计另一种结构 listpack。设计 listpack 的目的是取代 ziplist，但是 ziplist 使用范围比较广，替换起来比较复杂，所以目前只应用在新增加的 Stream 结构中。等到我们分析 Stream 时再讨论 listpack 的设计。由此可见，优秀的设计并不是一蹴而就的。

ziplist 提供常用函数如表 2-1 所示。

即使使用新的 listpack 格式，每插入一个新节点，也还可能需要进行两次内存拷贝。

（1）为整个链表分配新内存空间，主要是为新节点创建空间。

（2）将插入节点所有后驱节点后移，为插入节点腾出空间。

如果链表很长，则每次插入或删除节点时都需要进行大量的内存拷贝，这个性能是无法接受的，那么如何解决这个问题呢？这时就要用到 quicklist 了。

限于篇幅，关于 quicklist 内容，我们在下一文章分析。

本文内容摘自作者新书**《Redis 核心原理与实践》**，这本书深入地分析了 Redis 常用特性的内部机制与实现方式，大部分内容源自对 Redis 源码的分析，并从中总结出设计思路、实现原理。通过阅读本书，读者可以快速、轻松地了解 Redis 的内部运行机制。

经过该书编辑同意，我会继续在个人技术公众号（binecy）发布书中部分章节内容，作为书的预览内容，欢迎大家查阅，谢谢。

京东链接

豆瓣链接