自己建立网站教程_app在线设计_网站seo服务_营销推广案例

Redis 源码分析-内部数据结构 quicklist

quicklist 是 Redis 对外暴露的 list 数据结构的内部实现，经常被当作队列或栈使用，我们可以从常用的一些 api 上先思考一下它的结构

最常用的就是 lpush、lpop、rpush、rpop，同时它也支持 lindex 查询某元素在 list 中的索引，linsert 在指定元素旁边插入新元素。

从头、尾节点的 push、pop 来看，这就是双向链表最优秀的设计，提供头尾节点的 O(1) 复杂度操作，但在 ziplist 篇我提到了，普通的链表每个节点都需要一个前驱和后向指针，空间利用率低；且在链表元素过多时查询缓慢，ziplist 让每一个 entry 都存了上一个 entry 和自己的长度，用来实现双向遍历，节省了一些空间，但缺点是增加、修改元素时可能会带来更大的内存拷贝。

有没有办法进行一个时间和空间的权衡呢？quicklist 就是这样设计的

A doubly linked list of ziplists

一个 ziplist 的双向链表

那么每个 ziplist 存多少数据合适呢？比如我有 20 个元素，可以 4 个 ziplist，每个 ziplist 里 5 个元素；也可以 5 个 ziplist，每个 ziplist 里 4 个元素

• ziplist 的元素越多，进行操作时候需要内存拷贝的空间就越多，如果整个 quicklist 只有一个 ziplist 节点，那么其实就退化成了一个 ziplist 了
• ziplist 的元素越少，就需要更多的节点，空间利用率低，更容易产生内存碎片，如果每个 ziplist 只有一个元素，quicklist 就退化成一个普通的双向链表了

redis 提供了一个这样的配置项

list-max-ziplist-size // 默认-2

当取正值的时候，表示按照数据项个数来限定每个 quicklist 节点上的 ziplist 长度。比如，当这个参数配置成 5的时候，表示每个 quicklist 节点的 ziplist 最多包含 5 个数据项。

当取负值的时候，表示按照占用字节数来限定每个 quicklist 节点上的 ziplist 长度。这时，它只能取 -1 到 -5 这五个值，每个值含义如下：

• -5: 每个quicklist节点上的ziplist大小不能超过64 Kb
• -4: 每个quicklist节点上的ziplist大小不能超过32 Kb
• -3: 每个quicklist节点上的ziplist大小不能超过16 Kb
• -2: 每个quicklist节点上的ziplist大小不能超过8 Kb（Redis 默认值）
• -1: 每个quicklist节点上的ziplist大小不能超过4 Kb

quicklist

typedef struct quicklist {quicklistNode *head;quicklistNode *tail;unsigned long count;        /* total count of all entries in all ziplists */unsigned int len;           /* number of quicklistNodes */int fill : 16;              /* fill factor for individual nodes */unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist;

和我们认知里的双向链表一样：

• head 头节点指针
• tail 尾节点指针
• count 整个 quicklist 中的元素数量
• len ziplist 的数量
• fill: 16bit，ziplist大小设置，存放list-max-ziplist-size参数的值
• compress: 16bit，节点压缩深度设置，存放list-compress-depth参数的值

list-max-ziplist-size 的值在最上面已经解释过含义了，list-compress-depth是出于这样的考虑：

当列表很长的时候，最容易被访问的很可能是两端的数据，中间的数据被访问的频率比较低，所以 redis 提供了该选项选择压缩的中间数据，取值含义如下：

• 0: 是个特殊值，表示都不压缩。这是Redis的默认值。
• 1: 表示quicklist两端各有1个节点不压缩，中间的节点压缩。
• 2: 表示quicklist两端各有2个节点不压缩，中间的节点压缩。
• 3: 表示quicklist两端各有3个节点不压缩，中间的节点压缩。

注意，两个端点（头、尾）是不会压缩的

quicklistNode

quicklistNode 是每一个 quicklist 节点的数据结构

typedef struct quicklistNode {struct quicklistNode *prev;     // 前一个节点struct quicklistNode *next;     // 下一个节点unsigned char *zl;              // 数据指针。如果当前节点的数据没有压缩，那么它指向一个 ziplist 结构// 否则，它指向一个quicklistLZF 结构。unsigned int sz;                // zl 指向的 ziplist 的总大小（包括数据项和 zlbytes 等）unsigned int count: 16;         // ziplist 里面包含的数据项个数 2 字节unsigned int encoding: 2;       // 目前只有 1/2 取值 分别代表没压缩和 LZF 压缩unsigned int container: 2;      // 目前只有 2 取值，本意是取 1 代表节点直接存数据，取 2 代表使用 ziplist 存数据unsigned int recompress: 1;     // 当我们使用 index 等命令查看到了已经被压缩的数据，需要暂时解压，再设置标记为合适的时候压缩回去unsigned int attempted_compress: 1; // 自动化测试使用unsigned int extra: 10;         // 拓展字段，暂时没用上
} quicklistNode;// 一个被压缩过的 ziplist
typedef struct quicklistLZF {unsigned int sz;    // 压缩后的 ziplist 大小char compressed[];  // 柔性数组（flexible array member），存放压缩后的 ziplist 字节数组
} quicklistLZF;

以下是某一时刻一个 quicklist 的示意图：

list-max-ziplist-size 3
list-compress-depth 2

• 两端各有 2。橙黄色节点没有被压缩，对应 list-compress-depth 为 2
• 第二个节点和倒数第二个节点，都是元素数量为 3 的 ziplist，对应 list-max-ziplist-size

代码分析

1. 创建函数

quicklist *quicklistNew(int fill, int compress) {quicklist *quicklist = quicklistCreate();// 将两个 option 设置到 quicklist 中quicklistSetOptions(quicklist, fill, compress);return quicklist;
}quicklist *quicklistCreate(void) {struct quicklist *quicklist;quicklist = zmalloc(sizeof(*quicklist));// 初始化时 头尾节点都为 nullquicklist->head = quicklist->tail = NULL;quicklist->len = 0;quicklist->count = 0;quicklist->compress = 0;		// 压缩深度quicklist->fill = -2;			// 单个 ziplist 最多 8 字节return quicklist;
}

创建函数其实很简单，只是按照默认值初始化，并把配置项的值放入到 quicklist 对应的 compress、fill 字段上

其他操作复杂的原因在于，比如插入操作过程中要判断是否达到了配置的值，达到的话需要创建新的 node 节点（包括 ziplist ）；删除操作如果 ziplist 为空了，那么对应的头部或尾部节点也要删除，还可能涉及到里面节点的解压缩问题…

2. 在指定节点插入（lpush、rpush 最终都会调用到这个函数）

// 在 quicklist 的指定位置插入元素，根据 after 判断在前还是在后
REDIS_STATIC void __quicklistInsertNode(quicklist *quicklist,quicklistNode *old_node,quicklistNode *new_node, int after) {if (after) {new_node->prev = old_node;if (old_node) {new_node->next = old_node->next;if (old_node->next)old_node->next->prev = new_node;old_node->next = new_node;}// 如果插入在之后且 old_node 是尾节点,更新尾节点if (quicklist->tail == old_node)quicklist->tail = new_node;} else {new_node->next = old_node;if (old_node) {new_node->prev = old_node->prev;if (old_node->prev)old_node->prev->next = new_node;old_node->prev = new_node;}// 如果插入在之前且 old_node 是头节点,更新头节点if (quicklist->head == old_node)quicklist->head = new_node;}// 为空时初始化头尾指针if (quicklist->len == 0) {quicklist->head = quicklist->tail = new_node;}if (old_node)quicklistCompress(quicklist, old_node);quicklist->len++;
}

3. push 操作

在调用这个函数前，会根据操作是 lpush 还是 rpush 确定 where 是头还是尾

void quicklistPush(quicklist *quicklist, void *value, const size_t sz, int where) {if (where == QUICKLIST_HEAD) {quicklistPushHead(quicklist, value, sz);} else if (where == QUICKLIST_TAIL) {quicklistPushTail(quicklist, value, sz);}
}

头插

int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {quicklistNode *orig_head = quicklist->head;// 可以直接插入if (likely(_quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {// 当前节点插入到头节点对应的 ziplist 的头部quicklist->head->zl =ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);// 更新新节点指向的 ziplist 的总大小quicklistNodeUpdateSz(quicklist->head);} else {// 创建新的 node 节点quicklistNode *node = quicklistCreateNode();// 创建对应的 ziplist，并把新元素插入到头node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);// 更新新节点指向的 ziplist 的总大小quicklistNodeUpdateSz(node);// 把当前节点插入到原头节点的前面_quicklistInsertNodeBefore(quicklist, quicklist->head, node);}// 总数据项 + 1quicklist->count++;// 头节点的里的元素数量 + 1quicklist->head->count++;return (orig_head != quicklist->head);
}REDIS_STATIC void _quicklistInsertNodeBefore(quicklist *quicklist,quicklistNode *old_node,quicklistNode *new_node) {__quicklistInsertNode(quicklist, old_node, new_node, 0);
}

尾插

int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {quicklistNode *orig_tail = quicklist->tail;// 是否可以直接插入if (likely(_quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) {quicklist->tail->zl =ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL);quicklistNodeUpdateSz(quicklist->tail);} else {quicklistNode *node = quicklistCreateNode();node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL);quicklistNodeUpdateSz(node);// 把当前节点插入到原尾节点的后面_quicklistInsertNodeAfter(quicklist, quicklist->tail, node);}// 更新当前 quicklist 中的元素总数和尾节点对应 ziplist 中的元素数量quicklist->count++;quicklist->tail->count++;return (orig_tail != quicklist->tail);
}REDIS_STATIC void _quicklistInsertNodeAfter(quicklist *quicklist,quicklistNode *old_node,quicklistNode *new_node) {__quicklistInsertNode(quicklist, old_node, new_node, 1);
}

lpush 和 rpush 的逻辑几乎是一样的，而对其进行压缩，是在 __quicklistInsertNode 函数的 quicklistCompress 函数中…