[Redis]1-高效的数据结构P2-Set

按照惯例，先丢一个官网文档链接。

上篇我们已经了解了高效的数据结构P1-String与Hash。
这篇，我们继续来了解Redis的 Set 与 Sorted set。

有序集合 Sorted set

Redis 有序集合是一组唯一的字符串（成员）集合，这些字符串根据一个关联的分数进行排序。

这种有序、元素唯一且根据关联的分数进行排序的高效操作的数据结构，简称ZSET。
可以用于：

动态排序：比如排行榜，每个元素可以代表一个实体（如用户、商品），分数表示排序依据（如积分、销量）。由于ZSET自动维护排序，你可以轻松获取排名靠前的成员、某个成员的排名，或者按分数范围查询。

# 添加
> zadd prices 8 sandwich 
(integer) 1

> zadd prices 100000 car 
(integer) 1

> zadd prices 6300 iphone 8900 iphonepro
(integer) 2

# 结果展示
> zrange prices 0 9 withscores
1) "sandwich"
2) "8"
3) "iphone"
4) "6300"
5) "iphonepro"
6) "8900"
7) "car"
8) "100000"

速率限制器：ZSET可以实现一种基于滑动窗口的速率限制器，利用时间戳作为分数，成员记录请求标识，自动移除过期的请求。

底层实现

ZSET通过包含跳表和哈希表的二端口数据结构实现。每个ZSET对象包含一个哈希表和一个跳表，成员和分数在两边各存一份，哈希表存member -> score，跳表存score -> member的排序关系。

typedef struct zset {
    dict *dict;
    zskiplist *zsl;
} zset;

其中哈希结构上章已经了解过了。ZSET的唯一性便是通过Hash Table实现的。总体结构也同Hash类似。

跳表用于维护成员的按分数排序，支持高效的插入、删除、排名查询和范围查询。
本篇进行跳表skiplist的介绍，并了解skiplist是如何设计以支持排序的。

源码地址点这，其结构由redis.h/zskiplistNode和redis.h/zskiplist两个结构定义。

zskiplist和zskiplistNode结构如下：

typedef struct zskiplist {
    struct zskiplistNode *header, *tail; //头、尾节点
    unsigned long length; // 长度
    int level;//记录目前跳跃表内，层数最大的那个节点的层数（表头节点的层数不计算在内）
} zskiplist;

typedef struct zskiplistNode {
    sds ele;              // 成员
    double score;         // 分数
    struct zskiplistNode *backward;// 后退指针
    struct zskiplistLevel {
        struct zskiplistNode *forward;// 前进指针
        unsigned long span;// 跨度，记录跳过的节点数（前进指针所指向节点和当前节点的距离）
    } level[];
};

zskiplistNode用于表示跳跃表节点，zskiplist结构用于保存跳跃表节点的相关信息。
zskiplistNode点的level数组可以包含多个元素，每个元素都包含一个指向其他节点的指针，程序可以通过这些层来加快访问其他节点的速度，一般来说，层的数量越多，访问其他节点的速度就越快。

较传统Node与List，zskiplistNode多了一个level[]结构，这是一个动态数组。每个元素表示该节点在某一层级的前进指针（指向下一个节点）和跨度（span，表示跳过的节点数）

    struct zskiplistLevel {
        struct zskiplistNode *forward;// 前进指针
        unsigned long span;// 跨度，记录跳过的节点数（前进指针所指向节点和当前节点的距离）
    } level[];

Redis zskiplistNode这个设计通过level[]存储的多层索引预计算节点关系（预存关系），让查找、插入和删除的复杂度从传统链表的O(N)降到O(log N)，接近二分查找的效率。
跳表的高层索引节点稀疏，低层节点密集，类似二分搜索的层次划分，快速定位目标节点或分数范围。

level[]有点类似闭包表的核心表设计，存储节点关系。

![[Pasted image 20250417234839.png]]

核心方法
阅读Redis的zskiplist.c，重点看zslInsert（插入）和zslGetRank（排名计算），理解level[]和span的实现。
zslInsert源码如下：

/* 插入一个新节点到跳表中，返回新插入的节点指针
 * 参数：
 *   zsl: 跳表对象
 *   score: 新节点的分数
 *   ele: 新节点的成员（字符串）
 * 返回：
 *   新插入的节点指针
 */
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x; // update数组记录每层插入位置的前驱节点
    unsigned long rank[ZSKIPLIST_MAXLEVEL];       // rank数组记录每层累积的跨度
    int i, level;

    serverAssert(!isnan(score)); // 确保分数不是NaN

    // 步骤1：查找插入位置，记录前驱节点和跨度
    x = zsl->header; // 从头节点开始
    for (i = zsl->level-1; i >= 0; i--) { // 从最高层逐层向下查找
        // 初始化rank，继承上一层的跨度（若非最高层）
        rank[i] = i == (zsl->level-1) ? 0 : rank[i+1];
        // 沿当前层前进，直到遇到分数更大或字典序更大的节点
        while (x->level[i].forward &&
               (x->level[i].forward->score < score ||
                (x->level[i].forward->score == score &&
                 sdscmp(x->level[i].forward->ele, ele) < 0))) {
            // 累加跨度，记录跳过的节点数
            rank[i] += x->level[i].span;
            x = x->level[i].forward; // 前进到下一个节点
        }
        // 记录当前层的前驱节点
        update[i] = x;
    }

    // 步骤2：随机生成新节点的层级
    level = zslRandomLevel(); // 随机层级，概率递减（通常p=0.25）
    if (level > zsl->level) { // 如果新层级超过当前最大层级
        for (i = zsl->level; i < level; i++) {
            rank[i] = 0; // 新层级的rank初始化为0
            update[i] = zsl->header; // 新层级的前驱是头节点
            update[i]->level[i].span = zsl->length; // 跨度设为跳表总长度
        }
        zsl->level = level; // 更新跳表最大层级
    }

    // 步骤3：创建新节点
    x = zslCreateNode(level, score, ele); // 分配新节点内存，初始化分数和成员
    for (i = 0; i < level; i++) { // 为每层设置指针和跨度
        // 插入新节点：将新节点的forward指向前驱的forward
        x->level[i].forward = update[i]->level[i].forward;
        update[i]->level[i].forward = x; // 前驱指向新节点

        // 更新跨度：新节点的跨度 = 前驱原跨度 - 已跳过的节点数
        x->level[i].span = update[i]->level[i].span - (rank[0] - rank[i]);
        update[i]->level[i].span = (rank[0] - rank[i]) + 1; // 前驱的新跨度
    }

    // 步骤4：处理更高层的跨度
    for (i = level; i < zsl->level; i++) {
        update[i]->level[i].span++; // 未插入新节点的层，跨度+1
    }

    // 步骤5：设置后退指针
    x->backward = (update[0] == zsl->header) ? NULL : update[0];
    if (x->level[0].forward)
        x->level[0].forward->backward = x;
    else
        zsl->tail = x;

    // 步骤6：更新跳表元数据
    zsl->length++; // 跳表长度+1

    return x; // 返回新节点
}

集合 Set

Redis 集合是一个无序且唯一的字符串集合（成员）。您可以使用 Redis 集合高效地：

跟踪唯一的项目（例如，跟踪访问特定博客文章的所有唯一 IP 地址。唯一事件ID）
表示关系（例如，具有给定角色的所有用户的集合）
执行常见的集合操作，如交集、并集和差集

和Java的HashSet一样，非常适合删除重复数据的集合。

Redis Set的底层实现主要依赖两种数据结构：哈希表（Hash Table）和整数集合（Intset）。

其中哈希结构上章已经了解过了。唯一性便是通过Hash Table实现的。那么整数集合Intset是干什么的呢？
用来提供动态数据结构选择的。

当Set包含非整数成员（如字符串）或成员数量较多时，使用哈希表。
当Set的所有成员都是整数（支持int16、int32、int64），且成员数量较少时（受set-max-intset-entries配置控制，默认512），使用整数集合Intset。
数量超过阈值会转换成Hash Table，且Intset到哈希表的转换是单向的（不可逆），因为哈希表支持任意字符串，而Intset只支持整数。

即，Redis Set的底层数据结构会根据存储的成员类型和数量动态选择。
intset源码地址。
源码如下：

typedef struct intset {
	// 编码类型（INTSET_ENC_INT16/32/64）
    uint32_t encoding;
    // 数组长度
    uint32_t length;
    // 保存元素的数组
    int8_t contents[];
} intset;

Intset是一个紧凑的有序数组，存储整数值，自动选择最小编码类型（int16_t、int32_t、int64_t）以节省内存。
Intset有编码升级机制：
当插入的整数超出当前编码范围（如int16_t溢出），Intset自动升级到更高编码（如int32_t），并重新分配内存。
且Intset有序。Intset按数值大小排序，插入时使用二分查找定位。

Redis通过设计一种转换机制，使用Intset来专门优化存储小规模整数集合，达到节省内存（紧凑存储）的目的，提升内存效率，且支持快速二分查找，适合小集合的查询。