跳跃表是一种有序数据结构，它通过在每个节点中维持多个指向其他节点的指针，从而达到快速访问节点的目的。
跳跃表支持平均O(logN)、最坏O(N)复杂度的节点查找。
Redis只在两个地方用到了跳跃表，一个是实现有序集合集，另一个是在集群节点中用作内部数据结构。
1. 定义
Redis的跳跃表由redis.h/zskiplistNode和redis.h/zskiplist两个结构定义，其中zskiplistNode结构用于表示跳跃表节点，而zskiplist结构则用于保存跳跃表节点的相关信息，比如节点的数量，以及指向表头节点和表尾节点的指针等。
image
1.1 跳跃表节点
跳跃表节点的实现由redis.h/zskiplistNode结构定义
/*
* 跳跃表节点
*/
typedef struct zskiplistNode {
// 成员对象，各个节点中的o1、o2、o3是节点保存的成员对象
robj *obj;
// 分值，各个节点的1.0、2.0和3.0是分值，从小到大排列
double score;
// 后退指针，用BW表示，指向位于当前节点的前一个节点，用于从表尾向表头遍历
struct zskiplistNode *backward;
// 层
struct zskiplistLevel {
// 前进指针
struct zskiplistNode *forward;
// 跨度
unsigned int span;
} level[];
} zskiplistNode;
1.1.1 层
跳跃表节点的 1evel 数组可以包含多个元素，每个元素都包含一个指向其他节点的指针，程序可以通过这些层来加快访问其他节点的速度，一般来说，层的数量越多，访问其他节点的速度就越快。
每次创建一个新跳跃表节点的时候，程序都根据幂次定律（power law,
越大的数出现的概率越小）随机生成一个介于1和32之间的值作为level数组的大小，这个大小就是层的"高度"。
1.1.2 前进指针
每层都有一个指向表尾方向的前进指针(level[i].forward属性)，用于从表头向表尾方向访问节点。
1.1.3 跨度
层的跨度(level[i].span属性)用于记录两个节点之间的距离:
- 两个节点之间的跨度越大，它们相距的就越远
- 指向NULL的所有前进指针的跨度都为0，因为它们没有连向任何节点
1.1.4 后退指针
节点的后退指针（backward属性）用于从表尾向表头方向访问节点∶ 跟可以一次跳过多个节点的前进指针不同，因为每个节点只有一个后退指针，所以每次只能后退至前一个节点。
1.1.5 分值和成员
节点的分值（score 属性）是一个double类型的浮点数，跳跃表中的所有节点都按分值从小到大来排序。
节点的成员对象（obj属性）是一个指针，它指向一个字符串对象，而字符串对象则保存着一个 SDS 值。
1.2 跳跃表
但通过使用一个 zskiplist结构来持有这些节点，程序可以更方便地对整个跳跃表进行处理，比如快速访问跳跃表的表头节点和表尾节点，或者快速地获取跳跃表节点的数量等信息。
/*
* 跳跃表
*/
typedef struct zskiplist {
// 表头节点和表尾节点
struct zskiplistNode *header, *tail;
// 表中节点的数量
unsigned long length;
// 表中层数最大的节点的层数
int level;
} zskiplist;
2. 源码
2.1 zslCreate 创建一个新的跳跃表
/*
* 创建并返回一个新的跳跃表
*
* T = O(1)
*/
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;
}
/*
* 创建一个层数为 level 的跳跃表节点，
* 并将节点的成员对象设置为 obj ，分值设置为 score 。
*
* 返回值为新创建的跳跃表节点
*
* T = O(1)
*/
zskiplistNode *zslCreateNode(int level, double score, robj *obj) {
// 分配空间
zskiplistNode *zn = zmalloc(sizeof(*zn)+level*sizeof(struct zskiplistLevel));
// 设置属性
zn->score = score;
zn->obj = obj;
return zn;
}
2.2 查找过程
如图所示，我们要定位到那个紫色的 kv，需要从 header 的最高层开始遍历找到第一个节点（最后一个比「我」小的元素），然后从这个节点开始降一层再遍历找到第二个节点（最后一个比「我」小的元素），然后一直降到最底层进行遍历就找到了期望的节点（最底层的最后一个比我「小」的元素）。
2.2.1 zslGetRank 查找包含给定分值和成员对象的节点在跳跃表中的排位
/* Find the rank for an element by both score and key.
*
* 查找包含给定分值和成员对象的节点在跳跃表中的排位。
*
* Returns 0 when the element cannot be found, rank otherwise.
*
* 如果没有包含给定分值和成员对象的节点，返回 0 ，否则返回排位。
*
* Note that the rank is 1-based due to the span of zsl->header to the
* first element.
*
* 注意，因为跳跃表的表头也被计算在内，所以返回的排位以 1 为起始值。
*
* T_wrost = O(N), T_avg = O(log N)
*/
unsigned long zslGetRank(zskiplist *zsl, double score, robj *o) {
zskiplistNode *x;
unsigned long rank = 0;
int i;
// 遍历整个跳跃表
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
// 遍历节点并对比元素
while (x->level[i].forward &&
(x->level[i].forward->score < score ||
// 比对分值
(x->level[i].forward->score == score &&
// 比对成员对象
compareStringObjects(x->level[i].forward->obj,o) <= 0))) {
// 累积跨越的节点数量
rank += x->level[i].span;
// 沿着前进指针遍历跳跃表
x = x->level[i].forward;
}
/* x might be equal to zsl->header, so test if obj is non-NULL */
// 必须确保不仅分值相等，而且成员对象也要相等
// T = O(N)
if (x->obj && equalStringObjects(x->obj,o)) {
return rank;
}
}
// 没找到
return 0;
}
2.2.2 zslGetElementByRank 根据排位在跳跃表中查找元素
/* Finds an element by its rank. The rank argument needs to be 1-based.
*
* 根据排位在跳跃表中查找元素。排位的起始值为 1 。
*
* 成功查找返回相应的跳跃表节点，没找到则返回 NULL 。
*
* T_wrost = O(N), T_avg = O(log N)
*/
zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
zskiplistNode *x;
unsigned long traversed = 0;
int i;
// T_wrost = O(N), T_avg = O(log N)
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
// 遍历跳跃表并累积越过的节点数量
while (x->level[i].forward && (traversed + x->level[i].span) <= rank)
{
traversed += x->level[i].span;
x = x->level[i].forward;
}
// 如果越过的节点数量已经等于 rank
// 那么说明已经到达要找的节点
if (traversed == rank) {
return x;
}
}
// 没找到目标节点
return NULL;
}
2.3 插入过程
2.3.1 zslRandomLevel 随机层数
/* Returns a random level for the new skiplist node we are going to create.
*
* 返回一个随机值，用作新跳跃表节点的层数。
*
* The return value of this function is between 1 and ZSKIPLIST_MAXLEVEL
* (both inclusive), with a powerlaw-alike distribution where higher
* levels are less likely to be returned.
*
* 返回值介乎 1 和 ZSKIPLIST_MAXLEVEL 之间（包含 ZSKIPLIST_MAXLEVEL），
* 根据随机算法所使用的幂次定律，越大的值生成的几率越小。
*
* T = O(N)
*/
int zslRandomLevel(void) {
int level = 1;
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
2.3.2 zslInsert 插入
/*
* 创建一个成员为 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));
// 在各个层查找节点的插入位置
// T_wrost = O(N^2), T_avg = O(N log N)
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
/* store rank that is crossed to reach the insert position */
// 如果 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;
}
/* we assume the key is not already inside, since we allow duplicated
* scores, and the re-insertion of score and redis object should never
* happen since the caller of zslInsert() should test in the hash table
* if the element is already inside or not.
*
* zslInsert() 的调用者会确保同分值且同成员的元素不会出现，
* 所以这里不需要进一步进行检查，可以直接创建新元素。
*/
// 获取一个随机值作为新节点的层数
// 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;
}
2.4 zslDelete 删除
/* Delete an element with matching score/object from the skiplist.
*
* 从跳跃表 zsl 中删除包含给定节点 score 并且带有指定对象 obj 的节点。
*
* T_wrost = O(N^2), T_avg = O(N log N)
*/
int zslDelete(zskiplist *zsl, double score, robj *obj) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
int i;
// 遍历跳跃表，查找目标节点，并记录所有沿途节点
// T_wrost = O(N^2), T_avg = O(N log N)
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
// 遍历跳跃表的复杂度为 T_wrost = O(N), T_avg = O(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)))
// 沿着前进指针移动
x = x->level[i].forward;
// 记录沿途节点
update[i] = x;
}
/* We may have multiple elements with the same score, what we need
* is to find the element with both the right score and object.
*
* 检查找到的元素 x ，只有在它的分值和对象都相同时，才将它删除。
*/
x = x->level[0].forward;
if (x && score == x->score && equalStringObjects(x->obj,obj)) {
// T = O(1)
zslDeleteNode(zsl, x, update);
// T = O(1)
zslFreeNode(x);
return 1;
} else {
return 0; /* not found */
}
return 0; /* not found */
}
参考资料
- 《Redis设计与实现》
- 《Redis深度历险》
- Redis为什么用跳表而不用平衡树？- https://mp.weixin.qq.com/s?__biz=MzA4NTg1MjM0Mg==&mid=2657261425&idx=1&sn=d840079ea35875a8c8e02d9b3e44cf95&scene=21#wechat_redirect
- Redis源码注释版 https://github.com/huangz1990/redis-3.0-annotated
