Redis底层数据结构-Quicklist源码分析

简介

quicklist是Redis 3.2中引入的数据结构，其本质是一个双向链表，链表的节点类型是ziplist，当ziplist节点过多，quicklist会退化成双向链表，以提高效率。

数据结构

quicklistNode

该结构为节点类型。 - prev：指向前驱节点 - next：指向后继节点 - zl：指向对应的压缩列表 - sz：压缩列表的大小 - encoding：采用的编码方式，1为原生，2代表使用LZF进行压缩 - container：为节点指向的容器类型，1代表none，2代表ziplist存储数据 - recompress：代表这个节点是否是压缩节点，如果是，则使用压缩节点前先进行解压缩，使用后重新压缩 - attempted_compress：测试时使用

typedef struct quicklistNode {
    struct quicklistNode *prev;
    struct quicklistNode *next;
    unsigned char *zl;
    unsigned int sz;             /* ziplist size in bytes */
    unsigned int count : 16;     /* count of items in ziplist */
    unsigned int encoding : 2;   /* RAW==1 or LZF==2 */
    unsigned int container : 2;  /* NONE==1 or ZIPLIST==2 */
    unsigned int recompress : 1; /* was this node previous compressed? */
    unsigned int attempted_compress : 1; /* node can't compress; too small */
    unsigned int extra : 10; /* more bits to steal for future usage */
} quicklistNode;

quicklistLZF

如果使用LZF算法进行压缩，节点指向的结构为quicklistLZF，其中sz表示compressed数组所占用字节大小。

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

## quicklist - head：指向头节点 - tail：指向尾节点 - count：quicklist中压缩列表的entry总数 - len：节点个数 - fill：每个节点的ziplist长度，可以通过参数list-max-ziplist-size配置节点所占内存大小 - compress：该值表示两端各有compress个节点不压缩

typedef struct quicklist {
    quicklistNode *head;
    quicklistNode *tail;
    unsigned long count;        /* total count of all entries in all ziplists */
    unsigned long 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;

quicklistEntry

• quicklist：指向当前元素所在的quicklist
• node：指向当前元素所在的节点
• zi：当前元素所在的ziplist
• value：当前节点的字符串内容
• longval：当前节点的整型值
• sz：节点的大小
• offset：该节点相对于整个ziplist的偏移量，即相当于第几个entry

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

quicklistIter

用于quicklist遍历的迭代器。 - quicklist：指向对应的quicklist - current：指向元素所在的节点 - zi：指向元素所在的ziplist - offset：表示节点在ziplist中的偏移量 - direction：表示迭代器的方向

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

接口

quicklistPushHead

该接口在头部插入元素。传入参数分别是待插入的quicklist，插入的数据value，大小sz。返回指为0代表没有新建head节点，1代表新建了head节点 基本逻辑如下： - 查看quicklist的head节点是否允许插入，可以就利用ziplist的接口进行插入。 - 否则新建head节点进行插入。

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);
        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);
}

quicklistReplaceAtIndex

更改元素需要传入index，基本逻辑是： - 删除原有元素 - 插入新元素

int quicklistReplaceAtIndex(quicklist *quicklist, long index, void *data,
                            int sz) {
    quicklistEntry entry;
    //检查index位置的元素是否存在
    if (likely(quicklistIndex(quicklist, index, &entry))) {
        //删除
        entry.node->zl = ziplistDelete(entry.node->zl, &entry.zi);
        //添加
        entry.node->zl = ziplistInsert(entry.node->zl, entry.zi, data, sz);
        //更新sz字段
        quicklistNodeUpdateSz(entry.node);
        quicklistCompress(quicklist, entry.node);
        return 1;
    } else {
        return 0;
    }
}

quicklistIndex

该方法主要通过元素的下标查找对应元素，找到index对应数据所在的节点，再调用ziplist的ziplistGet得到Index对应的数据。

//idx为需要查找的下标, 结果写入entry，返回0代表没找到，1代表找到
int quicklistIndex(const quicklist *quicklist, const long long idx,
                   quicklistEntry *entry) {
    quicklistNode *n;
    unsigned long long accum = 0;
    unsigned long long index;
    //idx为负值，表示从尾部向头部的偏移量
    int forward = idx < 0 ? 0 : 1; /* < 0 -> reverse, 0+ -> forward */
    //初始化entry
    initEntry(entry);
    entry->quicklist = quicklist;

    if (!forward) {
        //idx小于0，将idx设置为(-idx) - 1，从尾部查找
        index = (-idx) - 1;
        n = quicklist->tail;
    } else {
        index = idx;
        n = quicklist->head;
    }

    if (index >= quicklist->count)
        return 0;
    //遍历Node节点，找到对应的index元素
    while (likely(n)) {
        if ((accum + n->count) > index) {
            break;
        } else {
            D("Skipping over (%p) %u at accum %lld", (void *)n, n->count,
              accum);
            accum += n->count;
            n = forward ? n->next : n->prev;
        }
    }

    if (!n)
        return 0;

    D("Found node: %p at accum %llu, idx %llu, sub+ %llu, sub- %llu", (void *)n,
      accum, index, index - accum, (-index) - 1 + accum);
    //计算index所在的ziplist的偏移量
    entry->node = n;
    if (forward) {
        /* forward = normal head-to-tail offset. */
        entry->offset = index - accum;
    } else {
        /* reverse = need negative offset for tail-to-head, so undo
         * the result of the original if (index < 0) above. */
        entry->offset = (-index) - 1 + accum;
    }

    quicklistDecompressNodeForUse(entry->node);
    entry->zi = ziplistIndex(entry->node->zl, entry->offset);
    //利用ziplistGet获取元素
    ziplistGet(entry->zi, &entry->value, &entry->sz, &entry->longval);
    /* The caller will use our result, so we don't re-compress here.
     * The caller can recompress or delete the node as needed. */
    return 1;
}

总结

本文主要介绍了Quicklist的基本数据结构，并对该结构的基本增删改查接口做了简要分析。