skiplist
skiplist 源码分析(跳表,ZSet 底层)
跳表解决的问题:给 ZSet 一个「按 score 有序、还能 O(logN) 做范围和排名查询」的结构。核心思路是在有序链表上叠多层稀疏索引,查找从高层往低层逐级逼近,用「掷骰子定层高」的概率均衡代替红黑树的旋转再平衡,实现只有几百行。
注意:本地仓是 unstable(2026-03),跳表在 8.x 之后被大改过——成员 sds 不再是节点里的一个指针,直接内嵌进节点这块内存;level[0].span 被挪用来存节点元信息;dict 里存的也不再是 member→score,改成直接存跳表节点指针。
网上教程和《Redis 设计与实现》讲的都是老版布局,下面是新版真实结构。
代码块收起展开
// 基于本地 Redis 仓 (unstable, 2026-03), src/server.h
#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^64 elements */
#define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */ // 每升一层概率 1/4:期望上层节点数是下层的 1/4,塔更矮更省内存
/* Node info placed in level[0].span since it's unused at level 0 (static assert verified) */
typedef struct zskiplistNodeInfo {
uint16_t sdsoffset; /* Offset from node start to sds data (after sds header) */ // 节点首地址到内嵌 sds 数据的偏移
uint8_t levels; /* Number of levels in this node (1-32) */ // 本节点层高;老版节点自己不知道层高,要用时得额外传
uint8_t reserved;
} zskiplistNodeInfo;
typedef struct zskiplistNode {
double score; // 排序主键;score 相同再比成员字典序,保证全序
struct zskiplistNode *backward; // 后退指针只有一根(第 0 层),够 ZREVRANGE 反向走就行,省 31 根指针
struct zskiplistLevel {
struct zskiplistNode *forward;
/* Span is the number of elements between this node and the next node at this level.
* At level 0, span is repurposed to store zskiplistNodeInfo for regular nodes, */
unsigned long span; // 跨度:到 forward 之间隔几个节点,ZRANK O(logN) 靠它;第 0 层跨度恒为 1,字段被挪用存 nodeInfo
} level[]; // 柔性数组,节点几层就分配几个
/* sds ele is embedded after level[] array (assist zslGetNodeElement(node) to access it) */
// 成员字符串直接内嵌在节点尾部:老版这里是 sds ele 指针,指向另一块独立分配的内存
} zskiplistNode;
typedef struct zskiplist {
struct zskiplistNode *header, *tail; // header 是哨兵,不存数据,固定 32 层
unsigned long length; // 节点数,不含 header
int level; // 当前实际最高层,查找从这层起步而不是从 32 层
size_t alloc_size; // 缓存总内存占用,MEMORY USAGE 不用遍历
} zskiplist;
typedef struct zset {
dict *dict; // 存跳表节点指针,按 member 查:ZSCORE 单点 O(1)
zskiplist *zsl; // 按 score 有序:范围/排名 O(logN)。双结构指向同一批节点,成员只存一份
} zset;核心操作都在 t_zset.c。第 0 层 span 被挪用后,所有 span 读写都要过一层封装:
代码块收起展开
// 基于本地 Redis 仓 (unstable, 2026-03), src/t_zset.c
static inline unsigned long zslGetNodeSpanAtLevel(zskiplistNode *x, int level) {
if (level > 0) return x->level[level].span;
return x->level[0].forward ? 1 : 0; // 第 0 层真实跨度必然是 1(尾节点 0),不用存,现场算
}
// ... zslSetNodeSpanAtLevel / zslIncrNodeSpanAtLevel / zslDecrNodeSpanAtLevel 同理:level 0 一律跳过不写
/* Get embedded sds from node. Uses the stored offset to directly access the sds data */
sds zslGetNodeElement(const zskiplistNode *node) {
zskiplistNodeInfo *info = zslGetNodeInfo(node);
debugServerAssert(info->sdsoffset != ZSL_OFFSET_NO_ELE); // header 无成员,用哨兵偏移标记
return (char*)node + info->sdsoffset; // 指针加偏移直达,不经过第二次内存跳转
}
/* Compare {score, ele} with node. Returns: 1=bigger 0=equal -1=smaller */
int zslCompareWithNode(double score, sds ele, const zskiplistNode *n) {
if (/*score < */ n == NULL) return -1; /* NULL is +infinity, comes after any real node */ // 把「走到本层末尾」统一成一次比较,循环条件不用单独判 NULL
if (score < n->score) return -1;
if (score > n->score) return 1;
/* Scores are equal, compare elements lexicographically */
return sdscmp(ele, zslGetNodeElement(n));
}
static int zslRandomLevel(void) {
static const int threshold = ZSKIPLIST_P*RAND_MAX;
int level = 1;
while (random() < threshold) // 每次 1/4 概率再升一层:概率均衡代替旋转再平衡,期望层高 O(logN)
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
/* Insert an already-created node, with its score, element set into the skiplist
* at the correct position. Updates all forward/backward pointers and spans. */
static void zslInsertNode(zskiplist *zsl, zskiplistNode *node) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL]; // 每层「插入点的前驱」,之后穿针引线全靠它
unsigned long rank[ZSKIPLIST_MAXLEVEL]; // 每层前驱的排名,用来拆 span
zskiplistNode *x;
int i, level;
double score = node->score;
sds ele = zslGetNodeElement(node);
level = zslGetNodeInfo(node)->levels;
serverAssert(!isnan(score)); // NaN 无法全序比较,进来直接断言死
/* Find the position where this node should be inserted */
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) { // 从最高层往下:每层向右走到「下一个就该更大」为止再降层,O(logN) 的来源
rank[i] = i == (zsl->level-1) ? 0 : rank[i+1]; // 下层排名接着上层累加,不用重算
while (zslCompareWithNode(score, ele, x->level[i].forward) > 0) {
rank[i] += zslGetNodeSpanAtLevel(x, i);
x = x->level[i].forward;
}
update[i] = x;
}
/* Update skiplist level if needed */
if (level > zsl->level) { // 新节点比整表还高:高出的层前驱只能是 header
for (i = zsl->level; i < level; i++) {
rank[i] = 0;
update[i] = zsl->header;
zslSetNodeSpanAtLevel(update[i], i, zsl->length); // header 在新层暂时跨过全表,下面会被拆
}
zsl->level = level;
zslGetNodeInfo(zsl->header)->levels = level;
}
/* Insert the node at the found position */
for (i = 0; i < level; i++) { // 每层做一次链表插入,同时把前驱的 span 拆成两段
node->level[i].forward = update[i]->level[i].forward;
update[i]->level[i].forward = node;
/* update span covered by update[i] as node is inserted here */
zslSetNodeSpanAtLevel(node, i, zslGetNodeSpanAtLevel(update[i], i) - (rank[0] - rank[i]));
zslSetNodeSpanAtLevel(update[i], i, (rank[0] - rank[i]) + 1); // rank[0]-rank[i] = 前驱到新节点的距离
}
/* increment span for untouched levels */
for (i = level; i < zsl->level; i++) {
zslIncrNodeSpanAtLevel(update[i], i, 1); // 新节点没够到的高层:跨越它的 span 集体 +1,否则 ZRANK 全错
}
/* Update backward pointers */
node->backward = (update[0] == zsl->header) ? NULL : update[0]; // backward 不指 header,反向遍历到 NULL 即停
if (node->level[0].forward)
node->level[0].forward->backward = node;
else
zsl->tail = node;
zsl->length++;
}
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
int level;
serverAssert(!isnan(score));
// ... 注释省略:调用方需先查 dict 保证元素不存在
level = zslRandomLevel();
zskiplistNode *node = zslCreateNode(zsl, level, score, ele); // 单块 zmalloc:节点头 + level[] + 内嵌 sds 一次分配
zslInsertNode(zsl, node);
return node;
}
/* Update the score of an element inside the sorted set skiplist. */
static void zslUpdateScore(zskiplist *zsl, zskiplistNode *node, double newscore) {
/* Fast path: if the node, after the score update, would be still exactly
* at the same position, we can just update the score without
* actually removing and re-inserting the element in the skiplist. */
if ((node->backward == NULL || node->backward->score < newscore) &&
(node->level[0].forward == NULL || node->level[0].forward->score > newscore))
{
node->score = newscore; // 没越过左右邻居就原地改,一根指针都不动;ZINCRBY 小步调分大多走这里
return;
}
// ... 慢路径省略:重找 update[],zslUnlinkNode 摘下 + 改 score + zslInsertNode 重挂,复用同一个节点
}
/* Find the rank for an element by both score and key.
* Note that the rank is 1-based due to the span of zsl->header to the
* first element. */
unsigned long zslGetRank(zskiplist *zsl, double score, sds ele) {
zskiplistNode *x;
unsigned long rank = 0;
int i;
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
while (zslCompareWithNode(score, ele, x->level[i].forward) >= 0) { // >=:相等也前进,这样 rank 会把目标自己数进去
rank += zslGetNodeSpanAtLevel(x, i); // 查找顺路累加 span 就是排名,ZRANK 不用回头再数一遍
x = x->level[i].forward;
}
if (x != zsl->header && zslCompareWithNode(score, ele, x) == 0) {
return rank;
}
}
return 0; // 0 表示没找到,所以排名设计成 1-based
}ZADD 的落点 zsetAdd,能看清 dict 和跳表怎么配合:
代码块收起展开
// 基于本地 Redis 仓 (unstable, 2026-03), src/t_zset.c
int zsetAdd(robj *zobj, double score, sds ele, int in_flags, int *out_flags, double *newscore) {
// ... 参数与 listpack 分支省略:元素少走 listpack;超过 zset-max-listpack-entries/value
// 时 zsetConvertAndExpand() 升级成 skiplist 编码,随后落进下面的分支
if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
zset *zs = zobj->ptr;
zskiplistNode *znode;
dictEntry *de;
dictEntryLink bucket, link;
link = dictFindLink(zs->dict, ele, &bucket); // 一次哈希查找同时拿到「存在与否」和「该插哪个桶」,新增路径省掉 dictAdd 的第二次查找
if (link != NULL) {
de = *link;
// ... NX / GT / LT 短路省略
znode = dictGetKey(de); // dict 的 key 就是跳表节点指针,score 直接从节点上读,dict 不存 value
curscore = znode->score;
// ... INCR 加分与 NaN 检查省略
/* Remove and re-insert when score changes. */
if (score != curscore) {
zslUpdateScore(zs->zsl, znode, score); // 节点指针不变,所以 dict 完全不用动
*out_flags |= ZADD_OUT_UPDATED;
}
return 1;
} else if (!xx) {
/* Element doesn't exist - create node with embedded sds and add to skiplist */
znode = zslInsert(zs->zsl, score, ele);
/* Add node pointer to dict using the bucket we already found */
dictSetKeyAtLink(zs->dict, znode, &bucket, 1); // 双结构各写一次,但成员字符串只存节点里那一份
*out_flags |= ZADD_OUT_ADDED;
if (newscore) *newscore = score;
return 1;
} else {
*out_flags |= ZADD_OUT_NOP;
return 1;
}
}
// ...
}原理串讲
拿一条 ZADD rank 99 tom 走通全链路。命令进 zaddGenericCommand,最终落到 zsetAdd。
集合还小的时候编码是 listpack,一旦条数或成员长度超过 zset-max-listpack-* 阈值,zsetConvertAndExpand 把它整体转成 skiplist 编码——此后 zobj->ptr 指向一个 zset 结构,里面是 dict 和 zskiplist 两套索引。
为什么要养两套结构?因为两类查询天然冲突:ZSCORE 按 member 查是点查,哈希 O(1) 最合适;ZRANGEBYSCORE/ZRANK 按 score 查是序查,必须有序结构。
任何单一结构都得牺牲一头,Redis 干脆两个都要,代价只是每个元素多一个 dictEntry——而且新版连这点代价都在压:dict 里存的直接是跳表节点指针(no_value dict),member 字符串只有节点里内嵌的那一份。
zsetAdd 先 dictFindLink 查重。假设 tom 是新成员,走 zslInsert:先 zslRandomLevel 掷骰子,每次 1/4 概率升一层,得到比如 3 层;
再 zslCreateNode 一次 zmalloc 把节点头、3 个 level、sds 头加 “tom” 全塞进同一块内存,并把层高和 sds 偏移打包成 zskiplistNodeInfo 写进 level[0].span。
为什么敢挪用这个字段?因为第 0 层是全量链表,任何节点到下一个节点的跨度必然是 1(尾节点是 0),这个值不用存,zslGetNodeSpanAtLevel 现场就能算出来——于是每个节点白捡 8 字节,正好放元信息。
为什么要把 sds 内嵌进节点?老版每个元素是「节点 + 独立 sds」两次分配,比较成员时要多跳一次指针;内嵌后分配减半、释放时 zslFreeNode 一次搞定,查找路径上 score 和成员数据大概率在同一条 cache line 附近,对这种指针跳来跳去、本来就 cache 不友好的结构是实打实的加速。
然后 zslInsertNode 干真正的插入。第一段循环从 zsl->level-1 层的 header 出发:在每一层沿 forward 向右走,zslCompareWithNode 判断下一个节点还小就继续,走不动了就把当前节点记进 update[i]、降一层接着走。
高层稀疏,一步跨掉一大段;低层稠密,负责精确定位——这就是 O(logN)。同时 rank[i] 沿途累加 span,记录每层前驱是全表第几个节点。
第二段把新节点在 0 到 level-1 每层做链表插入,并用 rank[0] - rank[i](前驱到新节点的距离)把前驱原来的 span 拆成两段。
最容易漏的是第三段:新节点没够到的那些高层,跨越这个位置的 span 要集体加 1,否则表长变了、高层跨度没变,ZRANK 从此全错。
为什么用掷骰子而不像红黑树那样严格再平衡?因为概率已经给出期望意义上的平衡(P=0.25 时期望层高约 1.33,查找期望 O(logN)),省掉了旋转逻辑和为旋转付出的复杂度——跳表全部核心代码几百行,红黑树光删除就够写几百行,而 Redis 还需要在这个结构上叠 span、backward、范围删除这些定制,简单结构才改得动。
回到 zsetAdd 的另一条路:tom 已存在、只是改分(或 ZINCRBY)。
这时拿着 dict 里的节点指针直接调 zslUpdateScore,它先看新 score 有没有越过 backward 和 level[0].forward 两个邻居——没越过就原地改一个 double 完事;越过了才走「摘下、改分、重挂」,且复用原节点,dict 里的指针依然有效。
查询侧 ZRANK 走 zslGetRank:和插入同样的下坡路径,区别是比较条件用 >=(相等也前进,把目标自己数进去),沿途累加 span,到达即得 1-based 排名,找不到返回 0。
设计取舍
- 跳表 vs 红黑树:复杂度同为 O(logN),选跳表是因为实现简单、好魔改(span/backward 都是外挂字段),且范围查询定位到起点后顺着第 0 层链表平推即可,树上还得中序回溯。
- P 取 0.25 而不是教科书的 0.5:平均每节点只有 1/(1-P)≈1.33 个 level,指针开销减半,查找常数略增,拿时间换内存。
- 双结构不是双倍内存:dict 与跳表共享节点,member 只存内嵌那一份;真正翻倍的只是索引项。
zslUpdateScore的快路径说明:跳表改 score 本质是「删 + 插」,只有不越过邻居时才能退化成原地写,别把 ZINCRBY 想成永远 O(1)。- 小集合根本不用跳表:128 个元素以内走 listpack 连续内存暴力扫,比一堆指针的跳表更省更快,
zset-max-listpack-entries就是这个开关。