redis 的 zset 使用了 2 种数据结构来实现 O(logN)) 的插入和删除操作，核心数据结构由 dict 和 跳表实现 ：

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

dict 将元素映射到分数，实现 O(1) 的成员查找，避免遍历 skiplist

zskiplist 按分数排序元素，支持范围查询

为了节省内存， dict 和跳表是共享同一个 SDS 字符串的

zset 编码

zset 支持 2 种编码：

OBJ_ENCODING_SKIPLIST：使用跳表+dict 的完整实现
OBJ_ENCODING_LISTPACK：当元素较少的时候使用紧凑的 listpack 编码

代码位置：src/t_zset.c:1224-1235

robj *zsetTypeCreate(size_t size_hint, size_t val_len_hint) {
    // zset_max_listpack_entries默认是128
    // zset_max_listpack_value默认是64
    if (size_hint <= server.zset_max_listpack_entries &&
        val_len_hint <= server.zset_max_listpack_value)
    {
        // 采用listpack编码
        return createZsetListpackObject();
    }
    
    // 跳表+dict的完整实现
    robj *zobj = createZsetObject();
    zset *zs = zobj->ptr;
    dictExpand(zs->dict, size_hint);
    return zobj;
}

当然，zset 也支持编码转换，listpack 和 skiplist 可以互转

代码位置： src/rdb.c:2138-2144

/* Convert *after* loading, since sorted sets are not stored ordered. */
if (zsetLength(o) <= server.zset_max_listpack_entries &&
    maxelelen <= server.zset_max_listpack_value &&
    lpSafeToAdd(NULL, totelelen))
{
    zsetConvert(o, OBJ_ENCODING_LISTPACK);
}

zset 编码转换时机

1. rdb 加载后

从 rdb 文件加载到 zset 后，会检查是否能转换成 listpack 编码。

转换条件：

元素数量 ≤ server.zset_max_listpack_entries (默认 128)
最大元素长度 ≤ server.zset_max_listpack_value （默认 64 字节）
listpack 总大小安全（确保 listpack 的总大小不会超过 1G，避免 listpack 头部的 total bytes 字段发生溢出）

src/rdb.c:2138-2144

/* Convert *after* loading, since sorted sets are not stored ordered. */
if (zsetLength(o) <= server.zset_max_listpack_entries &&
    maxelelen <= server.zset_max_listpack_value &&
    lpSafeToAdd(NULL, totelelen))
{
    zsetConvert(o, OBJ_ENCODING_LISTPACK);
}

2. 添加元素时

listpack 编码的 zset 添加元素，满足下面其中一个条件就会转换成 skiplist 编码：

当前元素数量+1 > server.zset_max_listpack_entries （默认是 128）
新元素长度 > server.zset_max_listpack_value （默认 64 字节）
listpack 无法安全添加新元素
src/t_zset.c:1462-1474

// zsetAdd函数

/* check if the element is too large or the list
 * becomes too long *before* executing zzlInsert. */
if (zzlLength(zobj->ptr)+1 > server.zset_max_listpack_entries ||
    sdslen(ele) > server.zset_max_listpack_value ||
    !lpSafeToAdd(zobj->ptr, sdslen(ele)))
{
    zsetConvertAndExpand(zobj, OBJ_ENCODING_SKIPLIST, zsetLength(zobj) + 1);
} else {
    zobj->ptr = zzlInsert(zobj->ptr,ele,score);
    if (newscore) *newscore = score;
    *out_flags |= ZADD_OUT_ADDED;
    return 1;
}

3.批量操作后的优化

在调用 集合操作命令后（比如 ZUNIONSTORE, ZINTERSTORE, ZDIFFSTORE），会调用 zsetConvertToListpackIfNeeded() 用于优化结果集的内存，此时可能会将 skiplist 编码转换为 listpack 编码

src/t_zset/c:1330-1343

/* Convert the sorted set object into a listpack if it is not already a listpack
 * and if the number of elements and the maximum element size and total elements size
 * are within the expected ranges. */
void zsetConvertToListpackIfNeeded(robj *zobj, size_t maxelelen, size_t totelelen) {
    if (zobj->encoding == OBJ_ENCODING_LISTPACK) return;
    zset *zset = zobj->ptr;

    if (zset->zsl->length <= server.zset_max_listpack_entries &&
        maxelelen <= server.zset_max_listpack_value &&
        lpSafeToAdd(NULL, totelelen))
    {
        // 转换成 listpack 编码
        zsetConvert(zobj,OBJ_ENCODING_LISTPACK);
    }
}

使用集合操作命令的时候，结果会存储在 skiplist 编码的 dstzset 中（因为需要排序和去重），如果结果集不会空，就会检查能否转换为 listpack 编码，如果结果集比较小就可以转换了，更加节省内存。

一些关于 zset 的疑问

为什么 zset 不用红黑🌲或者 B🌲

The Skip list | Hacker News
跳表指针虽然会比 B 树消耗更多内存，但可以通过调整节点拥有特定层数的概率参数，使得跳表比 B 树的内存占用更少
排序集合经常会被用于 ZRANGE、ZREVRANGE 的操作，以链表的方式遍历跳表，缓存局部性会比树好一些
跳表实现简单，调试方便（打印树结构都很麻烦了）

#redis #zset

zset实现