在RedisObject这一篇博客中,有介绍到list结构的底层编码类型有OBJ_ENCODING_QUICKLIST,当时就发现这个底层数据结构被我遗漏了。昨天花了点时间补了补这个知识,看完发现这货就跟STL中的deque的思想一样,顿时觉得又是一个实现超级繁琐但很实用的数据结构。今天就带大家一起来看看这个“二合一”的数据结构。
quicklist是Redis在3.2版本加入的新数据结构,其是list列表的底层数据结构。
quicklist简介
为什么说quicklist是“二合一”呢?如果你看过STL中的deque的实现,就会知道deque是由一个map中控器和一个数组组成的数据结构,它既具有链表头尾插入便捷的优点,又有数组连续内存存储,支持下标访问的优点。Redis中是采用sdlist和ziplist来实现quicklist的,其中sdlist充当map中控器的作用,ziplist充当占用连续内存空间数组的作用。quicklist本身是一个双向无环链表,它的每一个节点都是一个ziplist。为什么这么设计呢?
- 双向链表在插入节点上复杂度很低,但它的内存开销很大,每个节点的地址不连续,容易产生内存碎片。
- ziplist是存储在一段连续的内存上,存储效率高,但是它不利于修改操作,插入和删除数都很麻烦,复杂度高,而且其需要频繁的申请释放内存,特别是ziplist中数据较多的情况下,搬移内存数据太费时!
Redis综合了双向链表和ziplist的优点,设计了quicklist这个数据结构,使它作为list键的底层实现。接下来,就要考虑每一个ziplist中存放的元素个数。
- 如果每一个ziplist中的元素个数过少,内存碎片就会增多。可以按照极端情况双向链表来考虑。
- 如果每一个ziplist中的元素个数过多,那么ziplist分配大块连续内存空间的难度就增大,同样会影响效率。
Redis的配置文件中,给出了每个ziplist中的元素个数设定,考虑使用场景需求,我们可以选择不同的元素个数。该参数设置格式如下:
list-max-ziplist-size -2
后面的数字可正可负,正、负代表不同函数,其中,如果参数为正,表示按照数据项个数来限定每个节点中的元素个数,比如3代表每个节点中存放的元素个数不能超过3;反之,如果参数为负,表示按照字节数来限定每个节点中的元素个数,它只能取-1~-5这五个数,其含义如下:
- -1 每个节点的ziplist字节大小不能超过4kb
- -2 每个节点的ziplist字节大小不能超过8kb
- -3 每个节点的ziplist字节大小不能超过16kb
- -4 每个节点的ziplist字节大小不能超过32kb
- -5 每个节点的ziplist字节大小不能超过64kb
另外,在quicklist的源码中提到了一个LZF的压缩算法,该算法用于对quicklist的节点进行压缩操作。list的设计目的是能够存放很长的数据列表,当列表很长时,必然会占用很高的内存空间,且list中最容易访问的是两端的数据,中间的数据访问率较低,于是就可以从这个出发点来进一步节省内存用于其他操作。Redis提供了一下的配置参数,用于表示中间节点是否压缩。
list-compress-depth 0
参数list-compress-depth的取值和含义对应如下:
- 0 特殊值,表示不压缩
- 1 表示quicklist两端各有一个节点不压缩,中间的节点压缩
- 2 表示quicklist两端各有两个节点不压缩,中间的节点压缩
- 3 表示quicklist两端各有三个节点不压缩,中间的节点压缩
- 以此类推。
quicklist的数据结构
quicklist的数据结构定义在quicklist.c文件中。
typedef struct quicklist {
quicklistNode *head;
quicklistNode *tail;
unsigned long count;
unsigned int len;
int fill : 16;
unsigned int compress : 16;
} quicklist;
每个quicklist结构占用32个字节的空间,下面来看看quicklist节点的数据结构。
typedef struct quicklistNode {
struct quicklistNode *prev;
struct quicklistNode *next;
unsigned char *zl;
unsigned int sz;
unsigned int count : 16;
unsigned int encoding : 2;
unsigned int container : 2;
unsigned int recompress : 1;
unsigned int attempted_compress : 1;
unsigned int extra : 10;
} quicklistNode;
每个quicklistnode也占用32个字节,上面介绍了,每个节点的数据存放格式有ziplist和quicklistLZF,后者是一种采用LZF压缩算法压缩的数据结构,其定义如下:
typedef struct quicklistLZF {
unsigned int sz;
char compressed[];
} quicklistLZF;
分析到这里,quicklist的大体结构以及呈现在我们面前了,请看下图,是不是豁然开朗:
另外,quicklist还提供了迭代器结构以及指向ziplist中的节点结构
typedef struct quicklistIter {
const quicklist *quicklist;
quicklistNode *current;
unsigned char *zi;
long offset;
int direction;
} quicklistIter;
typedef struct quicklistEntry {
const quicklist *quicklist;
quicklistNode *node;
unsigned char *zi;
unsigned char *value;
long long longval;
unsigned int sz;
int offset;
} quicklistEntry;
quicklist基本接口
Redis为每一个底层数据结构都提供了丰富的接口,以供上层数据结构调用,quicklist也不例外。为了篇幅不会过长,本博客仅仅剖析一下部分关键的接口的实现。
创建quicklist及其节点
创建一个quicklist需要为其设定各种参数,其由quicklistCreate函数实现。
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;
return quicklist;
}
创建完quicklist,接下来就是创建一个quicklist节点。
REDIS_STATIC quicklistNode *quicklistCreateNode(void) {
quicklistNode *node;
node = zmalloc(sizeof(*node));
node->zl = NULL;
node->count = 0;
node->sz = 0;
node->next = node->prev = NULL;
node->encoding = QUICKLIST_NODE_ENCODING_RAW;
node->container = QUICKLIST_NODE_CONTAINER_ZIPLIST;
node->recompress = 0;
return node;
}
PUSH操作
quicklist最重要的操作就是首尾插入节点,此操作由quicklistPush函数实现。PUSH操作不管是头部还是尾部压入都包含两个步骤:
- 如果插入节点中的ziplist大小没有超过限制(list-max-ziplist-size),那么直接调用ziplistPush函数压入
- 如果插入节点中的ziplist大小超过了限制,则新建一个quicklist节点(自然会创建一个新的ziplist),新的数据项会压入到新的ziplist,新的quicklist节点插入到原有的quicklist上
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))) {
quicklist->head->zl =
ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
quicklistNodeUpdateSz(quicklist->head);
} else {
quicklistNode *node = quicklistCreateNode();
node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
quicklistNodeUpdateSz(node);
_quicklistInsertNodeBefore(quicklist, quicklist->head, node);
}
quicklist->count++;
quicklist->head->count++;
return (orig_head != quicklist->head);
}
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->count++;
quicklist->tail->count++;
return (orig_tail != quicklist->tail);
}
POP操作
与PUSH操作对应的是POP操作,POP操作可以弹出首尾节点。
int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *slong) {
unsigned char *vstr;
unsigned int vlen;
long long vlong;
if (quicklist->count == 0)
return 0;
int ret = quicklistPopCustom(quicklist, where, &vstr, &vlen, &vlong,
_quicklistSaver);
if (data)
*data = vstr;
if (slong)
*slong = vlong;
if (sz)
*sz = vlen;
return ret;
}
int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *sval,
void *(*saver)(unsigned char *data, unsigned int sz)) {
unsigned char *p;
unsigned char *vstr;
unsigned int vlen;
long long vlong;
int pos = (where == QUICKLIST_HEAD) ? 0 : -1;
if (quicklist->count == 0)
return 0;
if (data)
*data = NULL;
if (sz)
*sz = 0;
if (sval)
*sval = -123456789;
quicklistNode *node;
if (where == QUICKLIST_HEAD && quicklist->head) {
node = quicklist->head;
} else if (where == QUICKLIST_TAIL && quicklist->tail) {
node = quicklist->tail;
} else {
return 0;
}
p = ziplistIndex(node->zl, pos);
if (ziplistGet(p, &vstr, &vlen, &vlong)) {
if (vstr) {
if (data)
*data = saver(vstr, vlen);
if (sz)
*sz = vlen;
} else {
if (data)
*data = NULL;
if (sval)
*sval = vlong;
}
quicklistDelIndex(quicklist, node, &p);
return 1;
}
return 0;
}
REDIS_STATIC void *_quicklistSaver(unsigned char *data, unsigned int sz) {
unsigned char *vstr;
if (data) {
vstr = zmalloc(sz);
memcpy(vstr, data, sz);
return vstr;
}
return NULL;
}
其他接口函数
Redis关于quicklist还提供了很多接口函数,这里只罗列出接口,没有具体实现。
void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl);
quicklist *quicklistCreateFromZiplist(int fill, int compress,
unsigned char *zl);
void quicklistInsertAfter(quicklist *quicklist, quicklistEntry *node,
void *value, const size_t sz);
void quicklistInsertBefore(quicklist *quicklist, quicklistEntry *node,
void *value, const size_t sz);
void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry);
void quicklistRotate(quicklist *quicklist);
unsigned int quicklistCount(quicklist *ql);
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);
size_t quicklistGetLzf(const quicklistNode *node, void **data);
quicklist小结
quicklist将sdlist和ziplist两者的优点结合起来,在时间和空间上做了一个均衡,能较大程度上提高Redis的效率。压入和弹出操作的时间复杂度都很理想。在源码中,还给出了很多接口函数,有兴趣的读者可以去quicklist.c文件中查看,如果对博客中的讲述觉得有问题的,可以在下方留言,多多交流,互相学习!
本文内容由网友自发贡献,版权归原作者所有,本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容,请联系:hwhale#tublm.com(使用前将#替换为@)