乐趣区

关于java:Redis源码剖析之快速列表quicklist

作者：

jiezi

在

@TOC
何为quicklist，上次说到ziplist每次变更的工夫复杂度都十分高，因为必须要从新生成一个新的ziplist来作为更新后的list，如果一个list十分大且更新频繁，那就会给redis带来十分大的累赘。如何既保留ziplist的空间高效性，又能不让其更新简单度过高？ redis的作者给出的答案就是quicklist。

其实说白了就是把ziplist和一般的双向链表联合起来。每个双链表节点中保留一个ziplist，而后每个ziplist中存一批list中的数据(具体ziplist大小可配置)，这样既能够防止大量链表指针带来的内存耗费，也能够防止ziplist更新导致的大量性能损耗，将大的ziplist化整为零。

数据结构

quicklist

一图胜千言，咱们来看下一个理论的quicklist在内存中长啥样。

大抵介绍下上图中不同的节点，所有redis中list其实都是quicklist，所以像pop push等命令中的list参数都是quicklist。quicklist各字段及其含意如下。

typedef struct quicklist {
    quicklistNode *head;        /* 头结点 */
    quicklistNode *tail;        /* 尾结点 */
    unsigned long count;        /* 在所有的ziplist中的entry总数 */
    unsigned long len;          /* quicklist节点总数 */
    int fill : QL_FILL_BITS;              /* 16位，每个节点的最大容量 */
    unsigned int compress : QL_COMP_BITS; /* 16位，quicklist的压缩深度，0示意所有节点都不压缩，否则就示意从两端开始有多少个节点不压缩 */
    unsigned int bookmark_count: QL_BM_BITS;  /*4位，bookmarks数组的大小，bookmarks是一个可选字段，用来quicklist重新分配内存空间时应用，不应用时不占用空间*/
    quicklistBookmark bookmarks[];
} quicklist;

能够看出quicklist其实就是简略的双链表，但这里多进去几个字段，先重点介绍下compress。在上图中我用了两种不同色彩的节点，其中绿色是一般的ziplist节点，而红色是被压缩后的ziplist节点(LZF节点)，LZF是种无损压缩算法。redis为了节俭内存空间，会将quicklist的节点用LZF压缩后存储，但这里不是全副压缩，能够配置compress的值，compress为0示意所有节点都不压缩，否则就示意从两端开始有多少个节点不压缩，像我上图图示中，compress就是1，示意从两端开始，有1个节点不做LZF压缩。compress默认是0(不压缩)，具体能够依据你们业务理论应用场景去配置。

为什么不全副节点都压缩，而是流出compress这个可配置的口子呢？其实从统计而已，list两端的数据变更最为频繁，像lpush,rpush,lpop,rpop等命令都是在两端操作，如果频繁压缩或解压缩会代码不必要的性能损耗。从这里能够看出 redis其实并不是一味谋求性能，它也在致力缩小存储占用、在存储和性能之间做trade-off。

这里还有个fill字段，它的含意是每个quicknode的节点最大容量，不同的数值有不同的含意，默认是-2，当然也能够配置为其余数值，具体数值含意如下：

  • -1: 每个quicklistNode节点的ziplist所占字节数不能超过4kb。（倡议配置）
  • -2: 每个quicklistNode节点的ziplist所占字节数不能超过8kb。（默认配置&倡议配置）
  • -3: 每个quicklistNode节点的ziplist所占字节数不能超过16kb。
  • -4: 每个quicklistNode节点的ziplist所占字节数不能超过32kb。
  • -5: 每个quicklistNode节点的ziplist所占字节数不能超过64kb。
  • 任意负数: 示意：ziplist构造所最多蕴含的entry个数，最大为215215。

quicklistNode

quicklistNode就是双链表的节点封装了，除了前后节点的指针外，这里还蕴含一些本节点的其余信息。比方是否是LZF压缩的节点、ziplist相干信息…… 具体如下：

typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;           /* quicklist节点对应的ziplist */
    unsigned int sz;             /* ziplist的字节数 */
    unsigned int count : 16;     /* ziplist的item数*/
    unsigned int encoding : 2;   /* 数据类型，RAW==1 or LZF==2 */
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    unsigned int recompress : 1; /* 这个节点以前压缩过吗？ */
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    unsigned int extra : 10; /* 未应用到的10位 */
} quicklistNode;

从上文中咱们曾经理解了一个quicklist某个时刻在内存中的样子，接下咱们来看下它是如何在数据插入删除时变动的。

quicklist的操作

创立

/* 创立一个新的quicklist.
 * 应用quicklistRelease()开释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;
}

create就没啥好说的了，但这里须要揭示下，fill值默认是-2，也就是说每个quicklistNode中的ziplist最长是8k字节，能够更具本人业务需要调整具体配置。

头插和尾插

对于list而已，头部或者尾部插入是最常见的操作了，但其实头插和尾插还算是比较简单。

/* 在quicklist的头部插入一条数据  
 * 如果在已存在节点插入，返回0
 * 如果是在新的头结点插入，返回1 */
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);  // 在头结点对应的ziplist中插入 
        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);
}

/* 在quicklist的尾部插入一条数据  
 * 如果在已存在节点插入，返回0
 * 如果是在新的头结点插入，返回1 */
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);
}

头插和尾插都调用了_quicklistNodeAllowInsert先判断了是否能间接在以后头|尾节点能插入，如果能就直接插入到对应的ziplist里，否则就须要新建一个新节点再操作了。 还记得上文中咱们说的fill字段吗，_quicklistNodeAllowInsert其实就是依据fill的具体值来判断是否曾经超过最大容量。

特定地位插入

头插尾插比较简单，但quicklist在非头尾插入就比拟繁琐了，因为须要思考到插入地位、前节点、后节点的存储状况。

/* 在一个曾经存在的entry后面或者前面插入一个新的entry 
 * 如果after==1示意插入到前面，否则是插入到后面  */
REDIS_STATIC void _quicklistInsert(quicklist *quicklist, quicklistEntry *entry,
                                   void *value, const size_t sz, int after) {
    int full = 0, at_tail = 0, at_head = 0, full_next = 0, full_prev = 0;
    int fill = quicklist->fill;
    quicklistNode *node = entry->node;
    quicklistNode *new_node = NULL;

    if (!node) {
        /* 如果entry中未填node，则从新创立一个node并插入到quicklist中 */
        D("No node given!");
        new_node = quicklistCreateNode();
        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        __quicklistInsertNode(quicklist, NULL, new_node, after);
        new_node->count++;
        quicklist->count++;
        return;
    }

    /* 查看要插入的节点是否是满的 */
    if (!_quicklistNodeAllowInsert(node, fill, sz)) {
        D("Current node is full with count %d with requested fill %lu",
          node->count, fill);
        full = 1;
    }

    if (after && (entry->offset == node->count)) {
        D("At Tail of current ziplist");
        at_tail = 1;
        if (!_quicklistNodeAllowInsert(node->next, fill, sz)) {
            D("Next node is full too.");
            full_next = 1;
        }
    }

    if (!after && (entry->offset == 0)) {
        D("At Head");
        at_head = 1;
        if (!_quicklistNodeAllowInsert(node->prev, fill, sz)) {
            D("Prev node is full too.");
            full_prev = 1;
        }
    }

    /* 不确定把新元素插到哪 */
    if (!full && after) { // 如果以后节点不满，就直接插入 
        D("Not full, inserting after current position.");
        quicklistDecompressNodeForUse(node);
        unsigned char *next = ziplistNext(node->zl, entry->zi);
        if (next == NULL) {
            node->zl = ziplistPush(node->zl, value, sz, ZIPLIST_TAIL);
        } else {
            node->zl = ziplistInsert(node->zl, next, value, sz);
        }
        node->count++;
        quicklistNodeUpdateSz(node);
        quicklistRecompressOnly(quicklist, node);
    } else if (!full && !after) {
        D("Not full, inserting before current position.");
        quicklistDecompressNodeForUse(node);
        node->zl = ziplistInsert(node->zl, entry->zi, value, sz);
        node->count++;
        quicklistNodeUpdateSz(node);
        quicklistRecompressOnly(quicklist, node);
    } else if (full && at_tail && node->next && !full_next && after) {
        /* 如果以后节点是满的，要插入的地位是以后节点的尾部，且后一个节点有空间，那就插到后一个节点的头部。*/
        D("Full and tail, but next isn't full; inserting next node head");
        new_node = node->next;
        quicklistDecompressNodeForUse(new_node);
        new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_HEAD);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        quicklistRecompressOnly(quicklist, new_node);
    } else if (full && at_head && node->prev && !full_prev && !after) {
        /* 如果以后节点是满的，要插入的地位是以后节点的头部，且前一个节点有空间，那就插到前一个节点的尾部。  */
        D("Full and head, but prev isn't full, inserting prev node tail");
        new_node = node->prev;
        quicklistDecompressNodeForUse(new_node);
        new_node->zl = ziplistPush(new_node->zl, value, sz, ZIPLIST_TAIL);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        quicklistRecompressOnly(quicklist, new_node);
    } else if (full && ((at_tail && node->next && full_next && after) ||
                        (at_head && node->prev && full_prev && !after))) {
        /* 如果以后节点是满的，前后节点也都是满的，那就创立一个新的节点插进去 */
        D("\tprovisioning new node...");
        new_node = quicklistCreateNode();
        new_node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        __quicklistInsertNode(quicklist, node, new_node, after);
    } else if (full) {
        /* 否则，以后节点是满的，咱们须要把它决裂成两个新节点，个别用于插入到以后节点ziplist两头某个地位时 */
        D("\tsplitting node...");
        quicklistDecompressNodeForUse(node);
        new_node = _quicklistSplitNode(node, entry->offset, after);
        new_node->zl = ziplistPush(new_node->zl, value, sz,
                                   after ? ZIPLIST_HEAD : ZIPLIST_TAIL);
        new_node->count++;
        quicklistNodeUpdateSz(new_node);
        __quicklistInsertNode(quicklist, node, new_node, after);
        _quicklistMergeNodes(quicklist, node);
    }

    quicklist->count++;
}

代码比拟长，总结如下：

  • 如果以后被插入节点不满，直接插入。
  • 如果以后被插入节点是满的，要插入的地位是以后节点的尾部，且后一个节点有空间，那就插到后一个节点的头部。
  • 如果以后被插入节点是满的，要插入的地位是以后节点的头部，且前一个节点有空间，那就插到前一个节点的尾部。
  • 如果以后被插入节点是满的，前后节点也都是满的，要插入的地位是以后节点的头部或者尾部，那就创立一个新的节点插进去。
  • 否则，以后节点是满的，且要插入的地位在以后节点的两头地位，咱们须要把以后节点决裂成两个新节点，而后再插入。

数据删除

数据删除绝对于插入而言应该是反着来的，看完上面的代码后你就会发现不齐全是：

void quicklistDelEntry(quicklistIter *iter, quicklistEntry *entry) {
    quicklistNode *prev = entry->node->prev;
    quicklistNode *next = entry->node->next;
    int deleted_node = quicklistDelIndex((quicklist *)entry->quicklist,
                                         entry->node, &entry->zi);

    /* after delete, the zi is now invalid for any future usage. */
    iter->zi = NULL;

    /* If current node is deleted, we must update iterator node and offset. */
    if (deleted_node) {
        if (iter->direction == AL_START_HEAD) {
            iter->current = next;
            iter->offset = 0;
        } else if (iter->direction == AL_START_TAIL) {
            iter->current = prev;
            iter->offset = -1;
        }
    }
}
REDIS_STATIC int quicklistDelIndex(quicklist *quicklist, quicklistNode *node,
                                   unsigned char **p) {
    int gone = 0;

    node->zl = ziplistDelete(node->zl, p);
    node->count--;
    if (node->count == 0) {
        gone = 1;
        __quicklistDelNode(quicklist, node);
    } else {
        quicklistNodeUpdateSz(node);
    }
    quicklist->count--;
    /* If we deleted the node, the original node is no longer valid */
    return gone ? 1 : 0;
}

删除绝对于插入而言简略多了，我先看的插入逻辑，插入中有节点的决裂，但删除里却没有节点的合并，quicklist有节点最大容量，但没有最小容量限度。

其余API

了解了quicklist数据结构的设计，也根本就能猜测到每个api的具体实现了，这里我就不再列举代码了，有趣味能够自行查阅。

quicklist *quicklistCreate(void);  // 创立quicklist 
quicklist *quicklistNew(int fill, int compress);  // 用一些指定参数创立一个新的quicklist
void quicklistSetCompressDepth(quicklist *quicklist, int depth);  // 设置压缩深度 
void quicklistSetFill(quicklist *quicklist, int fill); // 设置容量下限 
void quicklistSetOptions(quicklist *quicklist, int fill, int depth); 
void quicklistRelease(quicklist *quicklist); // 开释quicklist
int quicklistPushHead(quicklist *quicklist, void *value, const size_t sz);  // 头部插入
int quicklistPushTail(quicklist *quicklist, void *value, const size_t sz);  // 尾部插入
void quicklistPush(quicklist *quicklist, void *value, const size_t sz, 
                   int where); // 指定头部或者尾部插入  
void quicklistAppendZiplist(quicklist *quicklist, unsigned char *zl); // 把一个ziplist放到quicklist中
quicklist *quicklistAppendValuesFromZiplist(quicklist *quicklist,   
                                            unsigned char *zl); // 把ziplist中的所有数据放到quicklist中
quicklist *quicklistCreateFromZiplist(int fill, int compress,
                                      unsigned char *zl);  // 从ziplist生成一个quicklist  
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); // 数据删除 
int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
                            int sz);   // 数据替换 
int quicklistDelRange(quicklist *quicklist, const long start, const long stop);  // 范畴删除  
quicklistIter *quicklistGetIterator(const quicklist *quicklist, int direction);  // 迭代器 
quicklistIter *quicklistGetIteratorAtIdx(const quicklist *quicklist,
                                         int direction, const long long idx);  // 从指定地位开始的迭代器  
int quicklistNext(quicklistIter *iter, quicklistEntry *node);   // 迭代器下一个地位  
void quicklistReleaseIterator(quicklistIter *iter);  // 开释迭代器  
quicklist *quicklistDup(quicklist *orig);  // 去重  
int quicklistIndex(const quicklist *quicklist, const long long index,
                   quicklistEntry *entry);  // 找到entry的下标索引 
void quicklistRewind(quicklist *quicklist, quicklistIter *li);
void quicklistRewindTail(quicklist *quicklist, quicklistIter *li);
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)); 
int quicklistPop(quicklist *quicklist, int where, unsigned char **data, 
                 unsigned int *sz, long long *slong); // 数据pop 
unsigned long quicklistCount(const quicklist *ql);
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len); // 比拟大小  
size_t quicklistGetLzf(const quicklistNode *node, void **data);  // LZF节点

参考资料

  • men_wen Redis源码分析和正文（七）— 疾速列表(quicklist)
  • 张铁蕾 Redis外部数据结构详解(5)——quicklist

本文是Redis源码分析系列博文，同时也有与之对应的Redis中文正文版，有想深刻学习Redis的同学，欢送star和关注。
Redis中文注解版仓库：https://github.com/xindoo/Redis
Redis源码分析专栏：https://zxs.io/s/1h
如果感觉本文对你有用，欢送一键三连。

评论

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

这个站点使用 Akismet 来减少垃圾评论。了解你的评论数据如何被处理。

更多文章