Redis 5 数据结构（三）quicklist

分类：Redis| 发布时间：2019-04-18 20:23:00

概述

本文是 Redis 5 数据结构系列的第三篇，主要讲解 quicklist 的结构和实现细节。

quciklist 是从 3.2 版本加入 Redis 的，用于取代普通的双向链表结构。主要用于减少拥有大量节点的列表的内存使用量，具体可以看这里： https://github.com/antirez/redis/pull/2143

quicklist 和节点的实现

quicklist 的结构如下所示：

#define QUICKLIST_NOCOMPRESS 0

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* 所有的 ziplists 中的所有节点的总数 */
    unsigned long len;          /* quicklistNodes 的个数 */
    int fill : 16;              /* 每个 quicklistNodes 的 fill factor */
    unsigned int compress : 16; /* 0 表示不进行压缩，否则这个值表示，表的头和尾不压缩的节点的个数  */
} quicklist;

在 Redis 内部，当创建一个列表对象时，实际上会使用 quicklist 来保存，此时 quicklist 的 fill 和 compress 字段是通过选项 list-max-ziplist-size 和 list-compress-depth 来进行设置的。此选项的具体含义可以查看 redis.conf 的 ADVANCED CONFIG 部分，或者查看我的另一篇文章： Redis 配置详解（五）

quicklist 节点（quicklistNode）的结构如下：

#define QUICKLIST_NODE_ENCODING_RAW 1
#define QUICKLIST_NODE_ENCODING_LZF 2

typedef struct quicklistNode {
    struct quicklistNode *prev;  /* 指向上一个节点 */
    struct quicklistNode *next;  /* 指向下一个节点 */
    unsigned char *zl;           /* 当节点保存的是压缩 ziplist 时，指向 quicklistLZF，否则指向 ziplist */
    unsigned int sz;             /* ziplist 的大小（字节为单位） */
    unsigned int count : 16;     /* ziplist 的项目个数 */
    unsigned int encoding : 2;   /* RAW==1 或者 LZF==2 */
    unsigned int container : 2;  /* NONE==1 或者 ZIPLIST==2 */
    unsigned int recompress : 1; /* 这个节点是否之前是被压缩的 */
    unsigned int attempted_compress : 1; /* 节点太小，不压缩 */
    unsigned int extra : 10; /* 未使用，保留字段 */
} quicklistNode;

quicklistNode 是一个 32 字节的结构体（64 位系统下）。其中使用了位域来让结构保持在 32 字节。

count 16比特，最大 65536（ziplist 最大为 64K，由于 ziplist 中的每个项目最小要 2 字节，因此这个值实际上会小于 32k）
encoding 2比特，RAW = 1, LZF = 2
container 2比特，NONE = 1, ZIPLIST = 2
recompress 1比特，bool，1 表示此节点由于使用中而被临时解压
attempted_compress 1比特，bool，用于测试
extra 10比特，未使用，将来使用，同时保持结构体 4 字节对齐

当 quicklistNode 指向的 ziplist 被压缩的情况下，上述的 zl 成员会指向如下的 quicklistLZF 结构：

typedef struct quicklistLZF {
    unsigned int sz; /* LZF size in bytes*/
    char compressed[];
} quicklistLZF;

sz 为压缩后的 ziplist 的长度
compressed 为压缩后的 ziplist 的实际内容

注意：压缩前的长度保存在 quicklistNode 的 sz 成员中。

下面给出几个 quicklist 的例子。

下图是一个拥有三个 quicklistNode 无压缩的 quicklist 的例子：

下图是一个拥有三个 quicklistNode 压缩深度为 1 的 quicklist 的例子：

quicklist API

quicklist *quicklistCreate(void);

创建一个新的 quicklist。

quicklist *quicklistNew(int fill, int compress);

给定 fill 和 compress 创建一个新的 quicklist。

void quicklistSetCompressDepth(quicklist *quicklist, int depth);
void quicklistSetFill(quicklist *quicklist, int fill);
void quicklistSetOptions(quicklist *quicklist, int fill, int depth);

设置 quicklist 的 fill 和 depth（compress）。

void quicklistRelease(quicklist *quicklist);

释放整个 quicklist。

int quicklistPushHead(quicklist *quicklist, void *value, const size_t sz);

插入一个项目到 quicklist 的头。

int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz);

插入一个项目到 quicklist 的尾。

#define QUICKLIST_HEAD 0
#define QUICKLIST_TAIL -1

void quicklistPush(quicklist *quicklist, void *value, const size_t sz, int where);

根据 where 的值选择插入项目到 quicklist 的头或者尾。

void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl);

创建一个新的 quicklistNode，将其指向预先创建的 ziplist（zl）。然后将这个节点追加到 quicklist 的末尾。

这个函数通常用于将数据从 RDB 文件加载到内存的情况。

quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist, unsigned char *zl);

将 zl 指向的 ziplist 的所有项目一个一个地追加到 quicklist。

这个方法用于将之前保存在 RDB 文件里的 ziplists 保存到新创建的 quicklist 中（这个 quicklist 设置的 ziplist 的大小要比保存在 RDB 的 ziplist 要小）。

返回传进来的 quicklist，传进来的 ziplist 会在追加到 quicklist 后释放。

quicklist *quicklistCreateFromZiplist(int fill, int compress, unsigned char *zl);

根据传进来的 ziplist 创建 quicklist（可能会有多个 quicklistNode）。

返回新的 quicklist。释放传进来的 ziplist（zl）。

typedef struct quicklistEntry {
    const quicklist *quicklist;
    quicklistNode *node;
    unsigned char *zi;
    unsigned char *value;
    long long longval;
    unsigned int sz;
    int offset;
} quicklistEntry;

int quicklistIndex(const quicklist *quicklist, const long long index,
                   quicklistEntry *entry);

根据以 0 为开始的索引获取 ‘quicklistEntry' 信息。 index 大于等于 0 时表示从头往后找，当 index 小于 0 时表示从后往前找，比如 -1 表示最后一个项目，-2 表示倒数第二个项目等等。

返回值：

1 表示对应索引的项目
0 表示未找到对应索引的项目返回 quicklist 指定的项目的信息和内容。

quicklistEntry 的各字段含义如下：

quicklist 指向关联的 quicklist
node 指向相应的 quicklistNode
zi 指向 ziplist 中的具体项目
value，sz 和 longval 对应 ziplistGet 的后面三个参数
offset 为项目在 ziplist 的索引

void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *node,
                          void *value, const size_t sz);

插入到指定位置的后面

void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *node,
                           void *value, const size_t sz);

插入到指定位置的前面

#define AL_START_HEAD 0
#define AL_START_TAIL 1

typedef struct quicklistIter {
    const quicklist *quicklist;
    quicklistNode *current;
    unsigned char *zi;
    long offset; /* offset in current ziplist */
    int direction;
} quicklistIter;

quicklistIter *quicklistGetIterator(const quicklist *quicklist, int direction);

返回一个 quicklist 迭代器，创建迭代器后需要调用 quicklistNext 对 quicklist 进行迭代。

direction 为 AL_START_HEAD 表示从头往后迭代，AL_START_TAIL 表示从后往前迭代。

quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist,
                                         int direction, const long long idx);

和 quicklistGetIterator 类似，只是多了一个 idx 参数用于指示迭代器的起始位置。

int quicklistNext(quicklistIter *iter, quicklistEntry *node);

获取迭代其器的下一个项目。返回 1 表示获取成功，返回 0 表示到达列表的末尾，此时 node 的值是无效的。

注意：当你正在对一个 quicklist 进行迭代时，不能对 quicklist 进行插入操作。但是你可以使用 quicklistDelEntry() 函数删除列表中的项目。

如果你在迭代过程中进行了插入操作，在插入操作后你应该重新创建迭代器。

iter = quicklistGetIterator(quicklist,<direction>);
quicklistEntry entry;
while (quicklistNext(iter, &entry)) {
    if (entry.value)
         [[ use entry.value with entry.sz ]]
    else
         [[ use entry.longval ]]
}

上述是一个使用 quicklist 迭代器的典型例子。

void quicklistReleaseIterator(quicklistIter *iter);

删除迭代器，如果迭代器正指向一个合法的节点，此函数会对节点重新编码（检查此节点之前是否是压缩状态的，若是则将其压缩）。

void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry);

删除指定的项目，可以在迭代过程中使用。

int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
                            int sz);

将索引为 index 的项目替换为 data 和 sz 指向的数据。函数内部会使用 ziplistDelete 先删除已有项目再用 ziplistInsert 插入新的项目。

返回 1,表示替换成功
返回 0,表示替换失败，quicklist 未发生变化

int quicklistDelRange(quicklist *quicklist, const long start, const long count);

删除 quicklist 指定范围的项目。

返回 1，表示删除成功
返回 0，表示未删除（count 为负数等非法情况）

quicklist *quicklistDup(quicklist *orig);

复制一个 quicklist。

void quicklistRotate(quicklist *quicklist);

将 quicklist 末尾的一个项目移到列表开头。

int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                       unsigned int *sz, long long *sval,
                       void *(*saver)(unsigned char *data, unsigned int sz));

获取 quicklist 开头（where 为 QUICKLIST_HEAD）或者末尾（where 为 QUICKLIST_TAIL）的项目的内容（列表中的项目不会被删除，不要被函数名误导了），若项目类型是字节数组类型则会将得到的数据调用 "saver 回调函数" 然后将回调函数的返回值设置给 data。

返回 0 表示列表为空
返回 1 表示成功，如果 data 未设置表示值在 sval）

int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                 unsigned int *sz, long long *slong);

调用 quicklistPopCustom，若项目类型为字节数组则使用 zmalloc 复制一份副本返回给 data，调用者负责释放这个内存。

unsigned long quicklistCount(const quicklist *ql);

返回 quicklist 的项目总数（直接返回 quicklist 结构中的 count）

int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);

内部直接调用 ziplistCompare。

size_t quicklistGetLzf(const quicklistNode *node, void **data);

返回 quicklistNode 中的 LZF 数据，返回值为 LZF 数据的长度。

quicklistNode 的分裂与合并

当插入数据到 quicklist 中间某个节点的时候，若果此节点的 ziplist 已经达到限值，则会发生 quicklistNode 的分裂，分裂点在要插入的位置。

假设我们有如下 quicklist，插入位置为 p 指向的位置。

假设我们要插入的 ziplist 已经到达限值，节点需要分裂。

可以看到分裂是以插入点进行分割的，分裂后我们要插入的数据会插入分裂后的前一个节点的末尾

插入后，为了避免由于分裂导致 quicklist 的节点碎片化，会调用 _quicklistMergeNodes(...) 对节点进行合并。

以分裂后的前一个节点作为中心节点（下面简称为 center）尝试进行如下合并：

(center->prev->prev, center->prev)
(center->next, center->next->next)
(center->prev, center)
(center, center->next)

在节点的两两合并前会先检查合并后的节点是否符合 quicklist 的 fill 限制，若符合则进行合并，否则不合并。在我们的例子中，假设只有 (center->next, center->next->next) 这一步是符合合并条件的。合并后的 quicklist 如下所示：

总结

quicklist 不会对过小的 ziplist 进行压缩，这个值为 MIN_COMPRESS_BYTES，目前是 48
quicklist 在对 ziplist 进行压缩后，会对比压缩后的长度和未压缩的长度，若压缩后的长度 - 未压缩的长度 < 8（MIN_COMPRESS_IMPROVE），则使用未压缩的数据
quicklist 的头节点和尾节点在任何时候都不会被压缩，因此可以保证将数据插入列表的头或者尾是高效的。
插入一个数据到压缩的节点时，需要先对节点的 ziplist 整个进行解压，插入后再次进行压缩
对中间某个已满的节点进行插入操作会导致此节点（假设此节点为 node）的 ziplist 在插入点进行分裂，此时这个 ziplist 会变成两个 ziplist（新建一个 quicklistNode 保存多出来的 ziplist，假设此节点为 newnode），然后将数据插入 node 指向的 ziplist 的末尾，然后对 node 进行 _quicklistMergeNodes 操作

凌云的博客

行胜于言