目录
引子
跳跃列表（Skip List），简称跳表。在前面我写过的一篇讲解HBase MemStore的文章中曾经提到，MemStore（其实也包括BlockCache）的基础数据结构都是JUC包提供的并发跳表ConcurrentSkipListMap。当时我就立flag说跳表是个有意思的数据结构，会写文章专门分析它，那么今天就来兑现承诺吧。
认识跳表
跳表的提出
跳表首先由William Pugh在其1990年的论文《Skip lists: A probabilistic alternative to balanced trees》中提出。由该论文的题目可以知道两点：
- 跳表是概率型数据结构。
- 跳表是用来替代平衡树的数据结构。准确来说，是用来替代自平衡二叉查找树（self-balancing BST）的结构。
看官可能会觉得这两点与之前文章中讲过的布隆过滤器有些相似，但实际上它们的原理和用途还是很不相同的。
由二叉树回归链表
考虑在有序序列中查找某个特定元素的情境：
- 如果该序列用支持随机访问的线性结构（数组）存储，那么我们很容易地用二分查找来做。
- 但是考虑到增删效率和内存扩展性，很多时候要用不支持随机访问的线性结构（链表）存储，就只能从头遍历、逐个比对。
- 作为折衷，如果用二叉树结构（BST）存储，就可以不靠随机访问特性进行二分查找了。
我们知道，普通BST插入元素越有序效率越低，最坏情况会退化回链表。因此很多大佬提出了自平衡BST结构，使其在任何情况下的增删查操作都保持O(logn)的时间复杂度。自平衡BST的代表就是AVL树、Splay树、2-3树及其衍生出的红黑树。如果推广之，不限于二叉树的话，我们耳熟能详的B树和B+树也属此类，常用于文件系统和数据库。
自平衡BST显然很香，但是它仍然有一个不那么香的点：树的自平衡过程比较复杂，实现起来麻烦，在高并发的情况下，加锁也会带来可观的overhead。如AVL树需要LL、LR、RL、RR四种旋转操作保持平衡，红黑树则需要左旋、右旋和节点变色三种操作。下面的动图展示的就是AVL树在插入元素时的平衡过程。
那么，有没有简单点的、与自平衡BST效率相近的实现方法呢？答案就是跳表，并且它简单很多，下面就来看一看。
设计思想与查找流程
跳表就是如同下图一样的许多链表的集合。
跳表具有如下的性质：
- 由多层组成，最底层为第1层，次底层为第2层，以此类推。层数不会超过一个固定的最大值Lmax。
- 每层都是一个有头节点的有序链表，第1层的链表包含跳表中的所有元素。
- 如果某个元素在第k层出现，那么在第1~k-1层也必定都会出现，但会按一定的概率p在第k+1层出现。
很显然这是一种空间换时间的思路，与索引异曲同工。第k层可以视为第k-1级索引，用来加速查找。为了避免占用空间过多，第1层之上都不存储实际数据，只有指针（包含指向同层下一个元素的指针与同一个元素下层的指针）。
当查找元素时，会从最顶层链表的头节点开始遍历。以升序跳表为例，如果当前节点的下一个节点包含的值比目标元素值小，则继续向右查找。如果下一个节点的值比目标值大，就转到当前层的下一层去查找。重复向右和向下的操作，直到找到与目标值相等的元素为止。下图中的蓝色箭头标记出了查找元素21的步骤。
通过图示，我们也可以更加明白“跳表”这个名称的含义，因为查找过程确实是跳跃的，比线性查找省时。若要查找在高层存在的元素（如25），步数就会变得更少。当数据量越来越大时，这种结构的优势就更加明显了。
插入元素的概率性
前文已经说过，跳表第k层的元素会按一定的概率p在第k+1层出现，这种概率性就是在插入过程中实现的。
当按照上述查找流程找到新元素的插入位置之后，首先将其插入第1层。至于是否要插入第2，3，4...层，就需要用随机数等方法来确定。最通用的实现方法描述如下。
int randomizeLevel(double p, int lmax) {
int level = 1;
Random random = new Random();
while (random.nextDouble() < p && level < lmax) {
level++;
}
return level;
}
得到层数k之后，就将新元素插入跳表的第1~k层。由上面的逻辑可知，随着层数的增加，元素被插入高层的概率会指数级下降。
下面的动图示出以p=1/2概率在跳表中插入元素的过程。这种方法也被叫做“抛钢镚儿”（coin flip），直到抛出正面/反面就停止。
相对于插入而言，删除元素没有这么多弯弯绕，基本上就是正常的单链表删除逻辑，因此不再展开。
复杂度分析
只要我们找出平均查找长度，自然就可以知道跳表的平均时间复杂度了。为了便于分析，我们将查找元素的流程反过来，从目的元素向头节点看，也就是变成向上、向左的操作。由于随机层数的计算是相互独立的，因此这样做并无妨。
假设从层数为l的某节点x出发，距离最顶层有k层。如果x有第l+1层的指针就向上走，没有的话就向左走。它们的概率分别为p与1-p。借用William Pugh画好的图如下。
设C[k]为向上走k层要走过的路径长度期望值，就有：C[0] = 0 ; C[k] = (1-p)·(1+C[k]) + p·(1+C[k-1])，最终可得：C[k] = k/p。
接下来设跳表中包含n个元素，那么第1层就有n个元素，第2层平均有np个，第3层平均有np2个……依次类推。可知，跳表层数的期望值为log1/pn。结合上面得出的结论，跳表的平均查找长度就是：[log1/pn - 1] / p。
也就是说，跳表增删查的平均时间复杂度为O(logn)，达到了平衡BST的水准。当然它毕竟是不稳定的，如果运气奇差无比，总是无法建立起层级结构的话，最坏时间复杂度仍然会退化到O(n)级别，但这个概率是非常微小的了。
由上面的分析我们还可以知道，跳表中单个节点的层数为1的概率为1-p，层数为2的概率为p(1-p)，层数为3的概率为p2(1-p)……依次类推。所以，单个节点层数的期望为：(1-p) · ∑[k=1→+∞](kpk-1) = 1/(1-p)。这也是单个节点中指针数量的平均值，列表如下。
可见，p越小，节点的空间开销也就越小，但（正规化的）查找长度会相应越大。
Redis的跳表实现
跳表在很多框架中都有广泛的应用，除Java并发包及HBase之外，比较著名的是Redis和leveldb。之前一直读Java系的源码，有些腻烦了，并且我最近正好在研究一些Redis调优方面的事情，就干脆拿Redis 4.0.14的源码来讲讲吧。
从zset到zskiplist
跳表在Redis中称为zskiplist，是其提供的有序集合（sorted set/zset）类型底层的数据结构之一。zset的定义如下，位于server.h中。
typedef struct zset {
dict *dict;
zskiplist *zsl;
} zset;
可见，除了zskiplist之外，zset还使用了KV哈希表dict。Redis中有序集合的默认实现其实是更为普遍的ziplist（压缩双链表），但在redis.conf中有两个参数可以控制它转为zset实现。
zset-max-ziplist-entries 128
zset-max-ziplist-value 64
也就是说，当有序集合中的元素数超过zset-max-ziplist-entries时，或其中任意一个元素的数据长度超过zset-max-ziplist-value时，就由ziplist自动转化为zset。具体逻辑参见t_zset.c中的zsetConvert()函数，不再赘述。
扯远了，回来看看zskiplist，它的定义就在zset上面。
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;
zskiplist的节点定义是结构体zskiplistNode，其中有以下字段。
- obj：存放该节点的数据。
- score：数据对应的分数值，zset通过分数对数据升序排列。
- backward：指向链表上一个节点的指针，即后向指针。
- level[]：结构体zskiplistLevel的数组，表示跳表中的一层。每层又存放有两个字段：
- forward是指向链表下一个节点的指针，即前向指针。
- span表示这个前向指针跳跃过了多少个节点（不包括当前节点）。
zskiplist就是跳表本身，其中有以下字段。
- header、tail：头指针和尾指针。
- length：跳表的长度，不包括头指针。
- level：跳表的层数。
下图示出一个length=3，level=5的zskiplist。
可见，zskiplist的第1层是个双向链表，其他层仍然是单向链表，这样做是为了方便可能的逆向获取数据的需求。
另外，节点中还会保存前向指针跳过的节点数span，这是因为zset本身支持基于排名的操作，如zrevrank指令（由数据查询排名）、zrevrange指令（由排名范围查询数据）等。如果有span值的话，就可以方便地在查找过程中累积出排名了。
以上是zskiplist相对于前述的传统跳表的两点不同，并且都给我们带来了便利。下面我们来继续读代码，看看它的部分具体操作。
创建zskiplist
zslCreate()函数位于t_zset.c中。
zskiplist *zslCreate(void) {
int j;
zskiplist *zsl;
zsl = zmalloc(sizeof(*zsl));
zsl->level = 1;
zsl->length = 0;
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;
}
并不难懂，唯一需要注意的是常量ZSKIPLIST_MAXLEVEL，它定义了zskiplist的最大层数，值为32，这也是上节图中的节点最高只到L32的原因。
向zskiplist插入元素
插入元素靠的是zslInsert()函数，有点长。
zskiplistNode *zslInsert(zskiplist *zsl, double score, robj *obj) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
unsigned int rank[ZSKIPLIST_MAXLEVEL];
int i, level;
serverAssert(!isnan(score));
x = zsl->header;
for (i = zsl->level-1; i >= 0; i--) {
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 &&
compareStringObjects(x->level[i].forward->obj,obj) < 0))) {
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
level = zslRandomLevel();
if (level > zsl->level) {
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);
for (i = 0; i < level; i++) {
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;
}
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;
}
该方法涉及到的细节很多，其大致执行流程如下：
- 按照前面讲过的查找流程，找到合适的插入位置。注意zset允许分数score相同，这时会根据节点数据obj的字典序来排序。
- 调用zslRandomLevel()方法，随机出要插入的节点的层数。
- 调用zslCreateNode()方法，根据层数level、分数score和数据obj创建出新节点。
- 每层遍历，修改新节点以及其前后节点的前向指针forward和跳跃长度span，也要更新最底层的后向指针backward。
其中维护了两个数组update和rank。update数组用来记录每一层的最后一个分数小于待插入score的节点，也就是插入位置。rank数组用来记录上述插入位置的上一个节点的排名，以便于最后更新span值。
随机计算层数的zslRandomLevel()方法如下。
int zslRandomLevel(void) {
int level = 1;
while ((random()&0xFFFF) < (ZSKIPLIST_P * 0xFFFF))
level += 1;
return (level<ZSKIPLIST_MAXLEVEL) ? level : ZSKIPLIST_MAXLEVEL;
}
注意p值由ZSKIPLIST_P常量定义，值为0.25，即被插入到高层的概率为1/4。
下图示出插入元素的流程。
查询元素排名/获取排名对应元素
从zslGetRank()和zslGetElementByRank()这两个函数可以更明显地看出通过累积span字段的值获取到排名的操作。代码本身比较容易理解，如下所示。
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;
}
if (x->obj && equalStringObjects(x->obj,o)) {
return rank;
}
}
return 0;
}
zskiplistNode* zslGetElementByRank(zskiplist *zsl, unsigned long rank) {
zskiplistNode *x;
unsigned long traversed = 0;
int i;
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;
}
if (traversed == rank) {
return x;
}
}
return NULL;
}
Redis作者对采用跳表的解释
比起我多费口舌，不如来看看Salvatore Sanfillipo（@antirez）本人说的话。他多年之前在Hacker News的一篇帖子上解释了自己为什么要在Redis中用跳表而不是树，原文如下，浅显易懂，就不翻译了：
- They are not very memory intensive. It's up to you basically. Changing parameters about the probability of a node to have a given number of levels will make then less memory intensive than btrees.
- A sorted set is often target of many ZRANGE or ZREVRANGE operations, that is, traversing the skip list as a linked list. With this operation the cache locality of skip lists is at least as good as with other kind of balanced trees.
- They are simpler to implement, debug, and so forth. For instance thanks to the skip list simplicity I received a patch (already in Redis master) with augmented skip lists implementing ZRANK in O(log(N)). It required little changes to the code.
The End
谢谢食用~
网友评论