Redis源码阅读笔记之跳表（SkipList）
在Redis中，还有一种有趣的数据结构-------跳表。它提供平均O(logN)，最坏O(N)的时间复杂度来对数据进行遍历操作，效率基本和平衡树差不多，但是比平衡树要简单，它不像链表，字符串，字典一样在Redis中应用广泛，跳表在Redis中只有两个主要的用途：有序集合（Sorted Set）底层实现以及集群节点中用于内部数据结构。其中有序集合中跳表的定义如下：
/*
* 有序集合
*/
typedef struct zset {
// 字典，键为成员，值为分值
// 用于支持 O(1) 复杂度的按成员取分值操作
dict *dict;
// 跳跃表，按分值排序成员
// 用于支持平均复杂度为 O(log N) 的按分值定位成员操作
// 以及范围操作
zskiplist *zsl;
} zset;
对于跳表，Redis底层使用了两个结构来定义实现，一个是zskiplistNode跳表节点，另一个是zskiplist跳表的定义。其中源码如下(redis.h/)：
/*
* Redis 对象
*/
#define REDIS_LRU_BITS 24
#define REDIS_LRU_CLOCK_MAX ((1<<REDIS_LRU_BITS)-1) /* Max value of obj->lru */
#define REDIS_LRU_CLOCK_RESOLUTION 1000 /* LRU clock resolution in ms */
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;
/*
* 跳跃表节点
*/
typedef struct zskiplistNode {
// 成员对象
robj *obj;
// 分值
double score;
// 后退指针
struct zskiplistNode *backward;
// 层
struct zskiplistLevel {
// 前进指针
struct zskiplistNode *forward;
// 跨度
unsigned int span;
} level[];
} zskiplistNode;
/*
* 跳跃表
*/
typedef struct zskiplist {
// 表头节点和表尾节点
struct zskiplistNode *header, *tail;
// 表中节点的数量
unsigned long length;
// 表中层数最大的节点的层数
int level;
} zskiplist;
在跳表的每一个节点zskiplistNode中保存着该节点的分数score，成员对象obj。并且还有一个层级结构体数组用来保存每一个跳表节点的每一层，跳表正是因为level才能快速访问指定节点，对于跳表的每一层level有两个属性，前进指针（访问下一个节点的某一层）和跨度（即与指向下一个节点之间的距离）。
zskiplist 用来描述一个跳表链，它提供了访问跳表的头结点以及尾节点，还提供了跳表链的长度，可以在O(1)的时间复杂度内快速计算出链表的长度，还记录了在一个跳表中层数最大节点的层数。（其中记录的跳表的长度length以及level头结点不计算在内）。如下图为跳表的示意图：
需要注意的时，在跳表中的所有节点都是按照score进行排序的，它与传统的跳表不同的是，redis中的跳表支持相同score的节点插入，如果score相同，它会使用成员对象obj进行比较，obj在字典序中较小的排在跳表的前面（头结点方向）。redis中跳表支持从表头或者表尾来遍历跳表，头结点的backward指针以及尾节点的forward指针都指向NULL，因此遍历时遇到NULL表示遍历结束。
对于跳表的每一个节点的每一层level都会保存着访问下一个节点的跨度span，用于计算每一个节点在跳表的排位（rank），如上图中score=2.0，obj=o2的节点的跨度就为2（1+1）。
下面看看在有序集合中redis时如何创建跳表以及插入数据的。t_zset.c中部分源码如下：
创建跳表：
/*
* 创建一个层数为 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;
}
/*
* 创建并返回一个新的跳跃表
*
* 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;
}
redis在初始化一个跳表节点的时候会随机初始化一个32以内的数字作为新节点的level，源码如下：
/* 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=32 之间（包含 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;
}
下面是插入在一个跳表中插入一个节点：
/*
* 创建一个成员为 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;
}
其余源码读者可以自行查看，下面是部分SkipList API:
