Redis源码（1）基本数据结构（下）

Redis源码（1）基本数据结构2：https://developer.aliyun.com/article/1508219

结构定义

跳跃表包含头尾节点、节点数目以及最大的层数，定义如下：

typedef struct zskiplist {
    // 表头节点和表尾节点
    struct zskiplistNode *header, *tail;
    // 表中节点的数量
    unsigned long length;
    // 表中层数最大的节点的层数
    int level;
} zskiplist;

值得注意的是，在计算level/length时，头结点不计算在内。

其中的跳跃表节点定义如下：

typedef struct zskiplistNode {

    // 成员对象(存储的对象类型，见下)
    robj *obj;
    // 分值（按照这个大小，升序排列）
    double score;
    // 后退指针（用作链表的倒序遍历）
    struct zskiplistNode *backward;
    // 层
    struct zskiplistLevel {
        // 前进指针
        struct zskiplistNode *forward;
        // 跨度
        unsigned int span;
    } level[];
} zskiplistNode;

跳表每个节点都具有多个层标记，每一个层标记带有两个属性：前进指针和跨度，前进指针表示访问其之后（score比它大）的其他节点，跨度表示当前节点跟后续节点的距离。

跳跃表 level 层级完全是随机的（插入节点时会随机生成1-32的数字）。一般来说，层级越多，访问节点的速度越快。

在跳跃表中，节点按各自所保存的分值从小到大排列。

obj是指向一个字符串对象，而字符串对象则保存着一个SDS值。

注意：在同一个跳跃表中，各个节点保存的成员对象必须是唯一的，但是多个节点保存的分值却可以是相同的：分值相同的节点将按照成员对象的大小（见后）来进行排序，成员对象较小的节点会排在前面(靠近表头的方向)，而成员对象较大的节点则会排在后面(靠近表尾的方向)。

其中的Redis成员对象定义如下：

typedef struct redisObject {
    // 类型(日后会说)
    unsigned type:4;
    // 编码(日后会说)
    unsigned encoding:4;
    // 对象最后一次被访问的时间(日后会说)
    unsigned lru:REDIS_LRU_BITS; /* lru time (relative to server.lruclock) */
    // 引用计数(日后会说)
    int refcount;
    // 指向实际值的指针
    void *ptr;
} robj;

注：成员对象的有关知识，参考Redis源码阅读（三）。

查找、删除、添加节点

那么Redis节点添加/插入操作是怎样的呢？我们以一个实际例子，来展现在跳跃表中，究竟是以何种方式进行节点的查找、删除和添加的。

• 添加节点

假设一个跳跃表刚开始为空，那么其实它就是一个简单的空链表结构：

当需要插入一个key=3的节点，得到一个随机值level = 3（随机抛出来的），那么该节点就具有level3的属性，此时跳跃表的结构如下：

·插入key = 2，随机的level = 1，如下：

···

·插入key = 100， 随机的level=2，如下：

okay，以上我们就完成了整个跳跃表的插入过程，那如果我们想查找66这个值，如何进行？

跳跃表的查询是从顶层往下找，那么会先从第顶层（即最高的level）开始找，查找方式就是一个二分查找，**如过该层找不到指定值，**就会跳到下一层，继续遍历，直到找到对应节点。

• 查找具体过程如下：

（level3时）66比1大，那就往同层的右边走，5还是比66小，再往右走就倒表尾了，因此下沉，从5的level3下沉到level2。

（level2）时，5的右边100比66大，所以从5的level2下沉到level1。

(level1)时，5的右边66恰好等于，于是返回这个节点。

• 删除操作呢？

其实跟查找操作差不多，找到删除节点之后，进行单向链表的删除操作，唯一的区别就是需要在多个level进行列表节点的删除。

以上部分参考链接：

https://blog.csdn.net/weixin_41622183/article/details/91126155

接口API

总览

函数	作用	时间复杂度
zslCreate	创建一个新的跳跃表。

zslFree	释放给定跳跃表，以及表中包含的所有节点。	， N 为跳跃表的长度。
zslInsert	将包含给定成员和分值的新节点添加到跳跃表中。	平均 ， N 为跳跃表长度。

zslDelete	删除跳跃表中包含给定成员和分值的节点。	平均 ， N 为跳跃表长度。
zslGetRank	返回包含给定成员和分值的节点在跳跃表中的排位。	平均 ， N 为跳跃表长度。

zslGetElementByRank

返回跳跃表在给定排位上的节点。

平均 ， N 为跳跃表长度。

zslIsInRange

给定一个分值范围（range）， 比如 0 到 15 ， 20 到 28，诸如此类， 如果给定的分值范围包含在跳跃表的分值范围之内， 那么返回 1 ，否则返回 0 。

通过跳跃表的表头节点和表尾节点， 这个检测可以用 复杂度完成。

zslFirstInRange	给定一个分值范围， 返回跳跃表中第一个符合这个范围的节点。	平均 。 N 为跳跃表长度。
zslLastInRange	给定一个分值范围， 返回跳跃表中最后一个符合这个范围的节点。	平均 。 N 为跳跃表长度。

zslDeleteRangeByScore	给定一个分值范围， 删除跳跃表中所有在这个范围之内的节点。	， N 为被删除节点数量。
zslDeleteRangeByRank	给定一个排位范围， 删除跳跃表中所有在这个范围之内的节点。	， N 为被删除节点数量。

其中，Redis中的跳表最大层级及概率定义如下：

#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */
#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */

创建新跳跃表如下：

zskiplist *zslCreate(void) {
    int j;
    zskiplist *zsl;
    // 分配空间
    zsl = zmalloc(sizeof(*zsl));
    // 设置高度和起始层数
    zsl->level = 1;
    zsl->length = 0;

    // 初始化表头节点
    // T = O(1)
    zsl->header = zslCreateNode(ZSKIPLIST_MAXLEVEL,0,NULL);
    for (j = 0; j < ZSKIPLIST_MAXLEVEL; j++) {
        zsl->header->level[j].forward = NULL;
        zsl->header->level[j].span = 0;
    }
    zsl->header->backward = NULL;

    // 设置表尾
    zsl->tail = NULL;

    return zsl;
}

从上诉代码可以看出，跳跃表的表头节点跟其他的跳跃表节点一样，只不过忽略了BW指针和分值、成员对象等信息。

创建新节点如下（请仔细思考源码的实现）：

/*
 * 创建一个成员为 obj ，分值为 score 的新节点，
 * 并将这个新节点插入到跳跃表 zsl 中。
 * 函数的返回值为新节点。
 * T_wrost = O(N^2), T_avg = O(N log N)
 */
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
    unsigned int rank[ZSKIPLIST_MAXLEVEL];
    int i, level;
    redisAssert(!isnan(score));
    // 在各个层查找节点的插入位置
    x = zsl->header;
    for (i = zsl->level-1; i >= 0; i--) {
        // 如果 i 不是 zsl->level-1 层
        // 那么 i 层的起始 rank 值为 i+1 层的 rank 值
        // 各个层的 rank 值一层层累积
        // 最终 rank[0] 的值加一就是新节点的前置节点的排位
        // rank[0] 会在后面成为计算 span 值和 rank 值的基础
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 沿着前进指针遍历跳跃表
        // T_wrost = O(N^2), T_avg = O(N log N)
        while (x->level[i].forward &&
            (x->level[i].forward->score < score ||
                // 比对分值
                (x->level[i].forward->score == score &&
                // 比对成员， T = O(N)
                compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
            // 记录沿途跨越了多少个节点
            rank[i] += x->level[i].span;
            // 移动至下一指针
            x = x->level[i].forward;
        }
        // 记录将要和新节点相连接的节点
        update[i] = x;
    }
     /* 
     // zslInsert() 的调用者会确保同分值且同成员的元素不会出现，
     * 所以这里不需要进一步进行检查，可以直接创建新元素。
     */
    // 获取一个随机值作为新节点的层数level
    // T = O(N)
    level = zslRandomLevel();
    // 如果新节点的层数比表中其他节点的层数都要大
    // 那么初始化表头节点中未使用的层，并将它们记录到 update 数组中
    // 将来也指向新节点
    if (level > zsl->level) {
        // 初始化未使用层
        // T = O(1)
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0;
            update[i] = zsl->header;
            update[i]->level[i].span = zsl->length;
        }
        // 更新表中节点最大层数
        zsl->level = level;
    }
    // 创建新节点
    x = zslCreateNode(level,score,obj);
    // 将前面记录的指针指向新节点，并做相应的设置
    // T = O(1)
    for (i = 0; i < level; i++) {
        // 设置新节点的 forward 指针
        x->level[i].forward = update[i]->level[i].forward;
        // 将沿途记录的各个节点的 forward 指针指向新节点
        update[i]->level[i].forward = x;
        /* update span covered by update[i] as x is inserted here */
        // 计算新节点跨越的节点数量
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        // 更新新节点插入之后，沿途节点的 span 值
        // 其中的 +1 计算的是新节点
        update[i]->level[i].span = (rank[0] - rank[i]) + 1;
    }
    /* increment span for untouched levels */
    // 未接触的节点的 span 值也需要增一，这些节点直接从表头指向新节点
    // T = O(1)
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++;
    }
    // 设置新节点的后退指针
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;
    // 跳跃表的节点计数增一
    zsl->length++;
    return x;
}

其中计算节点随机level的函数如下：

int zslRandomLevel(void) {
    int level = 1;
    //ZSKIPLIST为0.25
    while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
        level += 1;

    return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}

返回值介乎 1 和 ZSKIPLIST_MAXLEVEL 之间（包含ZSKIPLIST_MAXLEVEL），根据随机算法所使用的幂次定律，越大的值生成的几率越小。

补充：有序集合

Redis的有序集合数据结构是结合字典跟跳跃表实现的，其定义如下：

/*
 * 有序集合
 */
typedef struct zset {

    // 字典，键为成员，值为分值
    // 用于支持 O(1) 复杂度的按成员取分值操作
    dict *dict;

    // 跳跃表，按分值排序成员
    // 用于支持平均复杂度为 O(log N) 的按分值定位成员操作
    // 以及范围操作
    zskiplist *zsl;

} zset;

ZSET同时使用两种数据结构来持有同一个元素，从而提供 O(log(N)) 复杂度的有序数据结构的插入和移除操作。哈希表将 Redis 对象映射到分值上。而跳跃表则将分值映射到 Redis 对象上，以跳跃表的视角来看，可以说 Redis 对象是根据分值来排序的。

在Redis源码阅读（3）中会对zset的进一步介绍。

HyperLogLog

源码文件：hyperloglog.c。

Redis 在 2.8.9 版本添加了 HyperLogLog 结构。HyperLogLog 是用来做基数统计的算法，优点是，在输入元素的数量或者体积非常非常大时，计算基数所需的空间总是固定的、并且是很小的。在 Redis 里面，每个 HyperLogLog 键只需要花费 12 KB 内存，就可以计算接近 2^64 个不同元素的基数。这和计算基数时，元素越多耗费内存就越多的集合形成鲜明对比。

但是，因为 HyperLogLog 只会根据输入元素来计算基数，而不会储存输入元素本身，所以 HyperLogLog 不能像集合那样，返回输入的各个元素。

PS：什么是基数？

比如数据集 {1, 3, 5, 7, 5, 7, 8}， 那么这个数据集的基数集为 {1, 3, 5 ,7, 8}, 基数(不重复元素)为5。 基数估计就是在误差可接受的范围内，快速计算基数。

算法细节

它内部维护了 16384 个桶（bucket）来记录各自桶的元素数量。当一个元素到来时，它会散列到其中一个桶，以一定的概率影响这个桶的计数值。因为是概率算法，所以单个桶的计数值并不准确，但是将所有的桶计数值进行调合均值累加起来，结果就会非常接近真实的计数值。

为了便于理解HyperLogLog算法，我们先简化它的计数逻辑。因为是去重计数，如果是准确的去重，肯定需要用到 set 集合，使用集合来记录所有的元素，然后获取集合大小就可以得到总的计数。因为元素特别多，单个集合会特别大，所以将集合打散成 16384 个小集合。当元素到来时，通过 hash 算法将这个元素分派到其中的一个小集合存储，同样的元素总是会散列到同样的小集合。这样总的计数就是所有小集合大小的总和。使用这种方式精确计数除了可以增加元素外，还可以减少元素。

但是，集合打散并没有什么明显好处，因为总的内存占用并没有减少。HyperLogLog算法中每个桶所占用的空间实际上只有 6 个 bit，这 6 个 bit 自然是无法容纳桶中所有元素的，它记录的是桶中元素数量的对数值。

对数？怎么突然提到对数了？等等···Hyperloglog，log···难道对数才是这个数据结构的灵魂？！

先想想：一个随机的整数值，这个整数的尾部有一个 0 的概率是 50%，要么是 0 要么是 1（这里说的是二进制）。同样，尾部有两个 0 的概率是 25%，有三个零的概率是 12.5%，以此类推，有 k 个 0 的概率是 2^(-k)。如果我们随机出了很多整数，整数的数量我们并不知道，但是我们记录了整数尾部连续 0 的最大数量 K。我们就可以通过这个 K 来近似推断出整数的数量，这个数量就是 2^K！！

当然结果是非常不准确的，因为可能接下来你随机了非常多的整数，但是末尾连续零的最大数量 K 没有变化，但是估计值还是 2^K。你也许会想到要是这个 K 是个浮点数就好了，每次随机一个新元素，它都可以稍微往上涨一点点，那么估计值应该会准确很多。

HyperLogLog通过分配 16384 个桶，然后对所有的桶的最大数量 K 进行调合平均来得到一个平均的末尾零最大数量 K# ，K# 是一个浮点数，使用平均后的 2^K# 来估计元素的总量相对而言就会准确很多。不过这只是简化算法，真实的算法还有很多修正因子，因为涉及到的数学理论知识过于繁多，这里就不再精确描述。

下面我们看看Redis HyperLogLog 算法的具体实现。我们知道一个HyperLogLog实际占用的空间大约是 13684 * 6bit / 8 = 12k 字节。但是在计数比较小的时候，大多数桶的计数值都是零。如果 12k 字节里面太多的字节都是零，那么这个空间是可以适当节约一下的。Redis 在计数值比较小的情况下采用了稀疏存储，稀疏存储的空间占用远远小于 12k 字节。相对于稀疏存储的就是密集存储，密集存储会恒定占用 12k 字节。

不论是稀疏存储还是密集存储，Redis 内部都是使用字符串位图来存储 HyperLogLog 所有桶的计数值。

密集存储

密集存储的结构非常简单，就是连续 16384 个 6bit 串成的字符串位图。

那么给定一个桶编号，如何获取它的 6bit 计数值呢？这 6bit 可能在一个字节内部，也可能会跨越字节边界。我们需要对这一个或者两个字节进行适当的移位拼接才可以得到计数值。

假设桶的编号为idx，这个 6bit 计数值的起始字节位置偏移用 offset_bytes表示，它在这个字节的起始比特位置偏移用 offset_bits 表示。我们有：

offset_bytes = (idx * 6) / 8
offset_bits = (idx * 6) % 8

需要注意的是字节位序是左边低位右边高位，而通常我们使用的字节都是左边高位右边低位，我们需要在脑海中进行倒置。

稀疏存储

稀疏存储适用于很多计数值都是零的情况。下图表示了一般稀疏存储计数值的状态。

当多个连续桶的计数值都是零时，Redis 使用了一个字节来表示接下来有多少个桶的计数值都是零：00xxxxxx。前缀两个零表示接下来的 6bit 整数值加 1 就是零值计数器的数量，注意这里要加 1 是因为数量如果为零是没有意义的。比如 00010101表示连续 22 个零值计数器。6bit 最多只能表示连续 64 个零值计数器，所以 Redis 又设计了连续多个多于 64 个的连续零值计数器，它使用两个字节来表示：01xxxxxx yyyyyyyy，后面的 14bit 可以表示最多连续 16384 个零值计数器。这意味着 HyperLogLog 数据结构中 16384 个桶的初始状态，所有的计数器都是零值，可以直接使用 2 个字节来表示。

稀疏存储的一般状态如下：

回答上面的问题，何时从稀疏到非稀疏转换？

当稀疏存储的某个计数值需要调整到大于 32 （如33）时，Redis 就会立即转换 HyperLogLog 的存储结构，将稀疏存储转换成密集存储。

或者，稀疏存储占用的总字节数超过 3000 字节，这个阈值可以通过 hll_sparse_max_bytes 参数进行调整。

以上两种情况发生一种，就会触发稀疏到密集的转换，且不可逆。

计数缓存

前面提到 HyperLogLog 表示的总计数值是由 16384 个桶的计数值进行调和平均后再基于因子修正公式计算得出来的。它需要遍历所有的桶进行计算才可以得到这个值，中间还涉及到很多浮点运算。这个计算量相对来说还是比较大的。

所以 Redis 使用了一个额外的字段来缓存总计数值，这个字段有 64bit，最高位如果为 1 表示该值是否已经过期，如果为 0， 那么剩下的 63bit 就是计数值。

当 HyperLogLog 中任意一个桶的计数值发生变化时，就会将计数缓存设为过期，但是不会立即触发计算（惰性计算）。而是要等到用户显示调用 pfcount 指令时才会触发重新计算刷新缓存。缓存刷新在密集存储时需要遍历 16384 个桶的计数值进行调和平均，但是稀疏存储时没有这么大的计算量。也就是说只有当计数值比较大时才可能产生较大的计算量。另一方面如果计数值比较大，那么大部分 pfadd 操作根本不会导致桶中的计数值发生变化。

这意味着在一个极具变化的 HLL 计数器中频繁调用 pfcount 指令可能会有少许性能问题。关于这个性能方面的担忧在 Redis 作者 antirez 的博客中也提到了。不过作者做了仔细的压力的测试，发现这是无需担心的，pfcount 指令的平均时间复杂度就是 O(1)。

API接口

HHL数据结构暂时只需要理解其原理，代码分析等待内存编码源码阅读完毕再看。（这一部分疲了，等看了编码再回来）

参考链接：

基本介绍

https://www.runoob.com/redis/redis-hyperloglog.html

文中具体算法细节转载自（强烈推荐）

101表示连续 22 个零值计数器。6bit 最多只能表示连续 64 个零值计数器，所以 Redis 又设计了连续多个多于 64 个的连续零值计数器，它使用两个字节来表示：01xxxxxx yyyyyyyy，后面的 14bit 可以表示最多连续 16384 个零值计数器。这意味着 HyperLogLog 数据结构中 16384 个桶的初始状态，所有的计数器都是零值，可以直接使用 2 个字节来表示。

稀疏存储的一般状态如下：

[外链图片转存中…(img-tJ2j365k-1618042034281)]

回答上面的问题，何时从稀疏到非稀疏转换？

当稀疏存储的某个计数值需要调整到大于 32 （如33）时，Redis 就会立即转换 HyperLogLog 的存储结构，将稀疏存储转换成密集存储。

或者，稀疏存储占用的总字节数超过 3000 字节，这个阈值可以通过 hll_sparse_max_bytes 参数进行调整。

以上两种情况发生一种，就会触发稀疏到密集的转换，且不可逆。

计数缓存

前面提到 HyperLogLog 表示的总计数值是由 16384 个桶的计数值进行调和平均后再基于因子修正公式计算得出来的。它需要遍历所有的桶进行计算才可以得到这个值，中间还涉及到很多浮点运算。这个计算量相对来说还是比较大的。

所以 Redis 使用了一个额外的字段来缓存总计数值，这个字段有 64bit，最高位如果为 1 表示该值是否已经过期，如果为 0， 那么剩下的 63bit 就是计数值。

当 HyperLogLog 中任意一个桶的计数值发生变化时，就会将计数缓存设为过期，但是不会立即触发计算（惰性计算）。而是要等到用户显示调用 pfcount 指令时才会触发重新计算刷新缓存。缓存刷新在密集存储时需要遍历 16384 个桶的计数值进行调和平均，但是稀疏存储时没有这么大的计算量。也就是说只有当计数值比较大时才可能产生较大的计算量。另一方面如果计数值比较大，那么大部分 pfadd 操作根本不会导致桶中的计数值发生变化。

这意味着在一个极具变化的 HLL 计数器中频繁调用 pfcount 指令可能会有少许性能问题。关于这个性能方面的担忧在 Redis 作者 antirez 的博客中也提到了。不过作者做了仔细的压力的测试，发现这是无需担心的，pfcount 指令的平均时间复杂度就是 O(1)。

API接口

HHL数据结构暂时只需要理解其原理，代码分析等待内存编码源码阅读完毕再看。（这一部分疲了，等看了编码再回来）

参考链接：

基本介绍

https://www.runoob.com/redis/redis-hyperloglog.html

文中具体算法细节转载自（强烈推荐）

https://cloud.tencent.com/developer/article/1349691