简单来说,Redis 7.0 以后的 quicklist 其实相当于是 linkedlist + listpack 的变形组合。listpack 的优势在于可以高效利用内存空间,缺陷是不适合在单个 listpack 中存放太多数据,为了优化这一问题,Redis 通过链表的形式将大量不连续的 listpack 串联起来,这种结构就是 quicklist 的基本框架。
结构
quicklistNode
相当于链表中的数据节点,基本结构定义在 https://github.com/redis/redis/blob/7.2.4/src/quicklist.h#L46:
typedef struct quicklistNode {
struct quicklistNode *prev;
struct quicklistNode *next;
unsigned char *entry;
// 指向的 listpack 占用的字节数
size_t sz;
// 指向的 listpack 中的 entry 数量
unsigned int count : 16;
// 是否压缩,1表示不压缩,2表示通过LZF压缩
unsigned int encoding : 2;
unsigned int container : 2; /* PLAIN==1 or PACKED==2 */
unsigned int recompress : 1; /* was this node previous compressed? */
unsigned int attempted_compress : 1; /* node can't compress; too small */
unsigned int dont_compress : 1; /* prevent compression of entry that will be used later */
unsigned int extra : 9; /* more bits to steal for future usage */
} quicklistNode;
快速列表的节点构成继承了链表节点的一些特性,区别的地方在于 quicklistNode 支持对指向的数据以 Redis 提供的 LZF 算法进行压缩,如果节点指向 quicklistNode->entry 被压缩过,那么 entry 属性会指向 quicklistLZF
结构体:
typedef struct quicklistLZF {
// 压缩后的字节数
size_t sz;
char compressed[];
} quicklistLZF;
没被压缩的节点,quicklistNode->entry 会指向一个正常 quicklistEntry
结构体:
typedef struct quicklistEntry {
const quicklist *quicklist;
quicklistNode *node;
unsigned char *zi;
unsigned char *value;
long long longval;
size_t sz;
int offset;
} quicklistEntry;
其中 qucklistEntry->zi 指向的是存储数据的 listpack 结构。
多个 quicklistNode 统一由 quicklist 结构体来管理:
typedef struct quicklist {
quicklistNode *head;
quicklistNode *tail;
// 全部节点持有的listpack中的数据节点总数
unsigned long count;
// quicklistNode数量
unsigned long len;
signed int fill : QL_FILL_BITS;
// 节点是否全部压缩
unsigned int compress : QL_COMP_BITS;
unsigned int bookmark_count: QL_BM_BITS;
quicklistBookmark bookmarks[];
} quicklist;
quicklist->fill 字段用来标识每个 quicklistNode 的最大容量(默认为-2),fill 字段为负数(-N)时候,表示每个 quicklistNode 持有的 listpack 所占字节数不超过 2^(N+1) KB 个字节,fill 值为任意正数时候表示 listpack 最多允许包含的 entry 数量。
quicklist 结构体还附带管理了一个 quicklistBookmark
结构体的列表:
typedef struct quicklistBookmark {
quicklistNode *node;
char *name;
} quicklistBookmark;
quicklistBookmark 通常在 quicklist 中持有的节点数量很多的时候启用,当列表中有成千上万个节点的时候,可以通过 quicklistBookmark 快速定位到特定的 quicklistNode,这样就不用去花费时间来遍历列表获取特定的节点,而通常相对了大量节点而言,少数 quicklistBookmark 占用的内存空间是可以忽略不计的。
操作
quicklist 的创建基本和链表初始化差不多,分配内存后初始化头尾指针即可:
quicklist *quicklistCreate(void) {
struct quicklist *quicklist;
quicklist = zmalloc(sizeof(*quicklist));
quicklist->head = quicklist->tail = NULL;
quicklist->len = 0;
quicklist->count = 0;
quicklist->compress = 0;
quicklist->fill = -2;
quicklist->bookmark_count = 0;
return quicklist;
}
插入节点的操作也是基本的链表插入行为,还是比较容易理解的:
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;
}
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;
}
if (quicklist->head == old_node)
quicklist->head = new_node;
}
/* If this insert creates the only element so far, initialize head/tail. */
if (quicklist->len == 0) {
quicklist->head = quicklist->tail = new_node;
}
/* Update len first, so in __quicklistCompress we know exactly len */
quicklist->len++;
if (old_node)
quicklistCompress(quicklist, old_node);
quicklistCompress(quicklist, new_node);
}
quicklistNode 节点的压缩行为主要通过 lzf_compress()
完成:
REDIS_STATIC int __quicklistCompressNode(quicklistNode *node) {
#ifdef REDIS_TEST
node->attempted_compress = 1;
#endif
if (node->dont_compress) return 0;
/* validate that the node is neither
* tail nor head (it has prev and next)*/
assert(node->prev && node->next);
node->recompress = 0;
/* Don't bother compressing small values */
if (node->sz < MIN_COMPRESS_BYTES)
return 0;
quicklistLZF *lzf = zmalloc(sizeof(*lzf) + node->sz);
/* Cancel if compression fails or doesn't compress small enough */
if (((lzf->sz = lzf_compress(node->entry, node->sz, lzf->compressed,
node->sz)) == 0) ||
lzf->sz + MIN_COMPRESS_IMPROVE >= node->sz) {
/* lzf_compress aborts/rejects compression if value not compressible. */
zfree(lzf);
return 0;
}
lzf = zrealloc(lzf, sizeof(*lzf) + lzf->sz);
zfree(node->entry);
node->entry = (unsigned char *)lzf;
node->encoding = QUICKLIST_NODE_ENCODING_LZF;
return 1;
}