渐进式解析 Redis 源码 - 列表类型 t_list

阅读量

列表类型 作为 Redis 最重要的数据类型之一，广泛使用在 各种列表业务、消息队列 等场景

实际前面我们一直没有提到一种 数据结构 快速列表 quicklist，就是为了等这一刻，因为 快速列表 是 列表类型的专属 数据结构，而且 其实现 基础为 压缩列表 ziplist 和 双端链表 ……

快速列表quicklist 结构

首先我们分析一下 双端链表 和 列表 这两种数据结构的情况

  • 双端链表 在优点在于 头尾增删数据的 复杂度都是 O(1) 的；但是 列表类型 并不只是 lpush、lpop、rpush、rpop 这些操作，其中有很多 lindex、linsert 命令，这些的操作都是 O(n)，而且 双端链表 的每个节点都是独立的内存块，地址不连续，节点多了就容易产生 内存碎片
  • 压缩列表 的优点在于 内存是连续的，存储效率很高；但是不利于修改效率太低，经常会引发内存 realloc，而且甚至引起连锁更新就是 O(n2)

于是这个时候 快速列表就出现了，集合 双端链表 和 压缩列表 两种数据结构 的优势，如果对 这两种数据结构还不太熟悉的可以参考 双端链表、压缩列表
quicklist 结构

通过上图我们看到又两个 数据结构 组成的 快速列表结构，但是 很快我们就发现了一个问题，每个 链表节点 存放多少个 压缩列表节点 合适呢，比如 列表类型数据中共有 20个元素，我们可以有很多种存储方案，20x1、10x2 、5x4、4x5、2x10、1x20

  • 每个链表节点下的 压缩列表 元素越少，那么内存碎片就越多，从而降低存储效率。极端情况是每个链表节点下 压缩列表 只包含一个元素，这就是一个普通的双端链表
  • 每个链表节点下的压缩列表元素越多，那么 压缩列表 分配大块连续内存空间的难度就越大，有可能有的时候内存里有相当多小块的空闲空间，这些小块空间加起来很大，但却找不到一块足够大的空闲空间分配给压缩列表的情况，同样降低存储效率。极端情况是整个 快速列表 只有一个节点，所有的元素都分配在这独有的一个节点的 压缩链表 里面，这就是一个普通的压缩列表了

在这种纠结的、需要平衡的情况下，Redis 提供了一个配置参数 list-max-ziplist-size，对应 server 设置是 server.list_max_ziplist_size，默认值 -2

  • 如果配置 >0，代表限定的 每个链表节点上 存储的 最大 压缩列表 长度
  • 如果配置 <0，则是根据占用的字节数来限定 每个链表节点上 压缩列表长度
    • -5 每个链表节点上 的 压缩列表大小 <= 64KB
    • -4 每个链表节点上 的 压缩列表大小 <= 32KB
    • -3 每个链表节点上 的 压缩列表大小 <= 16KB
    • -2 每个链表节点上 的 压缩列表大小 <= 8KB
    • -1 每个链表节点上 的 压缩列表大小 <= 4KB

而 列表类型 还有一个特性，就是 开头和结尾 的节点数据往往是最容易被访问的，所以针对这种 场景，Redis 提供了另外一个配置参数 list-compress-depth，对应 server 配置是 server.list_compress_depth，默认值 0，能把中间的 节点 通过 LZF (无损压缩算法) 进行压缩，大量节省空间，各种参数对应含义如下：

  • 0 是个特殊值，表示都不压缩
  • 1 表示 快速列表两端各有 1个节点 不压缩，中间的节点压缩
  • 2 表示 快速列表两端各有 2个节点 不压缩，中间的节点压缩
  • 3 表示 快速列表两端各有 3个节点 不压缩，中间的节点压缩
  • 依此类推…

数据结构代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
// 快速列表节点数据结构
typedef struct quicklistNode {
    // 上一个节点地址
    struct quicklistNode *prev;
    // 下一个节点地址
    struct quicklistNode *next;
    // 数据指针地址，如果当前节点已经压缩，则指向 一个 quicklistLZF 地址，如果 没有压缩 则指向 压缩列表 地址
    unsigned char *zl;
    // 数据指向的 压缩列表 大小
    unsigned int sz;
    // 压缩列表里面的 元素个数
    unsigned int count : 16;
    // 表示压缩列表是否压缩以及用了哪个压缩算法，目前只有两种取值：2表示被压缩了（而且用的是LZF压缩算法），1表示没有压缩
    unsigned int encoding : 2;
    // 预留字段，用来表明一个节点下面是直接存数据，还是使用压缩列表存数据，或者用其它的结构来存数据
    unsigned int container : 2;
    // 当我们使用类似lindex这样的命令查看了某一项本来压缩的数据时，需要把数据暂时解压，这时就设置recompress=1做一个标记，等有机会再把数据重新压缩
    unsigned int recompress : 1;
    // 自动化测试程序相关
    unsigned int attempted_compress : 1;
    // 扩展字段
    unsigned int extra : 10;
} quicklistNode;
// 压缩后 的 压缩列表 结构
typedef struct quicklistLZF {
    // 压缩后的压缩列表大小
    unsigned int sz;
    // 柔性数组，存放压缩后的压缩列表元素数组
    char compressed[];
} quicklistLZF;
// 快速列表数据结构
typedef struct quicklist {
    // 头节点
    quicklistNode *head;
    // 尾节点
    quicklistNode *tail;
    // 所有 压缩列表元素之和
    unsigned long count;
    // 节点个数
    unsigned long len;
    // 存放 list-max-ziplist-size 参数值
    int fill : 16;
    // 存放 list-compress-depth 参数值
    unsigned int compress : 16;
} quicklist;
// 快速列表的迭代器结构
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;

下面我们就 Redis 列表类型 一些典型的 API 进行解析~~~

源码解析

由于有些方法里的代码量比较大，我们这里按照 典型的代码片段进行解析，同志们可以根据文章提示的代码位置 和 代码里面的关键词 在源码中搜素，可能数据结构一些元素 看不太懂什么意思，没关系，先混个脸熟，后面看完回头再看过来就明白了

快速列表 及其 节点的 创建函数为 quicklistCreate() 和 quicklistCreateNode()，逻辑比较简单，主要为 属性的初始化

列表类型中 用的最多的就是 头尾插入，但是 他们使用的处理函数都是 quicklistPush()，然后再根据位置分别调用 头插函数quicklistPushHead()、尾插函数quicklistPushTail()

  • 如果插入节点中的 ziplist 大小没有超过限制（ list-max-ziplist-size ），那么直接调用 ziplistPush 函数压入
  • 如果插入节点中的 ziplist 大小超过了限制，则新建一个 quicklist 节点（自然会创建一个新的 ziplist ），新的数据项会压入到新的 ziplist，新的 quicklist 节点插入到原有的 quicklist 上

不过这里还有另外一个 函数 likely()，属于 系统优化函数，是 linux 提供给程序员的编译优化方法，目的是将 ‘分支转移’ 的信息提供给编译器，这样编译器可以对代码进行优化，以减少指令跳转带来的性能下降

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
/* quicklist.c: line 480 */
int quicklistPushHead(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_head = quicklist->head;
    // 节点没有满发生的概率比较大，也就是数据项直接插入到当前节点的可能性大
    if (likely(
            // 判断该头部节点是否允许插入，计算头部节点中的大小和fill参数设置的大小相比较
            _quicklistNodeAllowInsert(quicklist->head, quicklist->fill, sz))) {
        // 执行到此，说明允许插入，直接调用ziplistpush插入节点即可
        quicklist->head->zl =
            ziplistPush(quicklist->head->zl, value, sz, ZIPLIST_HEAD);
        // 更新头部大小
        quicklistNodeUpdateSz(quicklist->head);
    } else {
        // 执行到此，说明头部节点已经满了，需要重新创建一个节点
        quicklistNode *node = quicklistCreateNode();
        // 将新节点压入新创建的ziplist中，并与新创建的quicklist节点关联起来
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_HEAD);
        // 更新大小
        quicklistNodeUpdateSz(node);
        // 将新创建的quicklist节点关联到quicklist中
        _quicklistInsertNodeBefore(quicklist, quicklist->head, node);
    }
    // 更新total数据项个数
    quicklist->count++;
    // 更新头结点的数据项个数
    quicklist->head->count++;
    // 如果尾部quicklist节点指针没变，返回0，反之返回1
    return (orig_head != quicklist->head);
}
int quicklistPushTail(quicklist *quicklist, void *value, size_t sz) {
    quicklistNode *orig_tail = quicklist->tail;
    if (likely(
            // 判断该尾部节点是否允许插入，计算尾部节点中的大小和fill参数设置的大小相比较
            _quicklistNodeAllowInsert(quicklist->tail, quicklist->fill, sz))) {
        // 执行到此，说明允许插入，直接调用ziplistpush插入节点即可，将新数据项push到ziplist的尾部
        quicklist->tail->zl =
            ziplistPush(quicklist->tail->zl, value, sz, ZIPLIST_TAIL);
        // 更新尾部节点大小
        quicklistNodeUpdateSz(quicklist->tail);
    } else {
        // 执行到此，说明尾部节点已经满了，需要重新创建一个节点
        quicklistNode *node = quicklistCreateNode();
        // 创建一个新的ziplist，并将新数据项插入，然后与新创建的quicklist节点关联起来
        node->zl = ziplistPush(ziplistNew(), value, sz, ZIPLIST_TAIL);
        // 更新该quicklist节点的大小
        quicklistNodeUpdateSz(node);
        // 将新创建的quicklist与quicklist关联起来
        _quicklistInsertNodeAfter(quicklist, quicklist->tail, node);
    }
    // 更新quicklist的数据项个数
    quicklist->count++;
    // 更新尾部节点的数据项个数
    quicklist->tail->count++;
    // 如果尾部quicklist节点指针没变，返回0，反之返回1
    return (orig_tail != quicklist->tail);
}

列表另外一个常用的操作就是 POP，执行函数为 quicklistPop()，而具体逻辑处理函数为 quicklistPopCustom()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
/* quicklist.c: line 1385 */
// 如果弹出节点是字符串值，data，sz存放弹出节点的字符串值；如果弹出节点是整型值，slong存放弹出节点的整型值
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;
    // 调用底层实现函数，传入的 _quicklistSaver是一个函数指针，用于深拷贝节点的值
    int ret = quicklistPopCustom(quicklist, where, &vstr, &vlen, &vlong, _quicklistSaver);
    // 给data，sz，slong赋值
    if (data)
        *data = vstr;
    if (slong)
        *slong = vlong;
    if (sz)
        *sz = vlen;
    return ret;
}
// 如果弹出节点是字符串值，data，sz存放弹出节点的字符串值，如果弹出节点是整型值，slong存放弹出节点的整型值
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;
    // 获取quicklist节点
    quicklistNode *node;
    if (where == QUICKLIST_HEAD && quicklist->head) {
        node = quicklist->head;
    } else if (where == QUICKLIST_TAIL && quicklist->tail) {
        node = quicklist->tail;
    } else {
        return 0;
    }
    // 获取ziplist中的节点
    p = ziplistIndex(node->zl, pos);
    // 获取该节点的值
    if (ziplistGet(p, &vstr, &vlen, &vlong)) {
        // 如果是字符串值
        if (vstr) {
            if (data)
                // _quicklistSaver函数用于深拷贝取出返回值
                *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;
}

本文作者： wettper
本文链接： http://www.web-lovers.com/redis-source-list.html
版权声明： 本博客所有文章除特别声明外，均采用 CC BY-NC-SA 3.0 许可协议。转载请注明出处！

赏

谢谢你请我吃糖果

支付宝
微信

扫一扫，分享到微信

微信分享二维码

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

wettper <br /><br /> 现 离职状态 程序猿 <br /><br /> 叫醒我的不止阳光，还有梦想！<br /><br /> 喵喵、念念、健身、吉他！