概述
跳跃表是一个随机化的数据结构,可以看做是二叉树的一个变种,它在性能上和红黑树,AVL树不想上下。目前在Reids和lucene的倒排索引文件中都有应用(后文会介绍跳跃表在这两种情况下的应用)。
原理
跳表的原理非常简单,跳表其实就是一种可以进行二分查找的有序链表。跳表的数据结构模型如图:
可以看到,跳表在原有的有序链表上面增加了多级索引,通过索引来实现快速查找。首先在最高级索引上查找最后一个小于当前查找元素的位置,然后再跳到次高级索引继续查找,直到跳到最底层为止,这时候以及十分接近要查找的元素的位置了(如果查找元素存在的话)。由于根据索引可以一次跳过多个元素,所以跳查找的查找速度也就变快了
在有n个元素的普通有序链表里查询一个数据最坏情况下需要比较n次,因此时间复杂度为O(n)。如果在链中部节点加一个指针,则比较次数可以减少到n/2+1。搜索时,首先将欲搜索元素与中间元素进行比较。如果欲搜索的元素较小,则仅需搜索链表的左半部分,否则,只要在链表右半部分进行比较即可。由此可见,跳跃表的查询的时间复杂度为O(logn)。插入,删除也是寻找合适的位置的操作,因此时间复杂度均为O(logn),最坏情况下时间复杂度为O(n)。
层的分配
每一次插入一个新结点时,最好的做法就是根据当前表的结构得到一个合适的层数,插入后可以让跳跃表尽量接近理想的结构,但这在实现上会非常困难。Pugh 的论文中提出的方法是根据概率随机为新结点生成一个层数,具体的算法如下:
a.给定一个概率 p(p 小于 1),产生一个 [0,1) 之间的随机数;
b.如果这个随机数小于 p,则层数加 1;
c.重复以上动作,直到随机数大于概率 p(或层数大于程序给定的最大层数限制)。
完整代码:
#include <iostream>
#include <cstdlib>
#include <cstring>
#define P 0.25
#define MAX_LEVEL 32
using namespace std;
struct Node
{
int key;
Node ** forward;
Node(int key = 0, int level = MAX_LEVEL)
{
this->key = key;
forward = new Node*[level];
memset(forward, 0, level * sizeof(Node*));
}
};
class SkipList
{
private:
Node * header;
int level;
private:
int random_level();
public:
SkipList();
~SkipList();
bool insert(int key);
bool find(int key);
bool erase(int key);
void print();
};
//确定层数
int SkipList::random_level()
{
int level = 1;
while ((rand() & 0xffff) < (P * 0xffff) && level < MAX_LEVEL)
{
level++;
}
return level;
}
SkipList::SkipList()
{
header = new Node;
level = 0;
}
SkipList::~SkipList()
{
Node * cur = header;
Node * next = nullptr;
while (cur)
{
next = cur->forward[0];
delete cur;
cur = next;
}
header = nullptr;
}
bool SkipList::insert(int key)
{
Node * node = header;
Node * update[MAX_LEVEL];
memset(update, 0, MAX_LEVEL * sizeof(Node*));
for (int i = level - 1; i >= 0; i--)
{
while (node->forward[i] && node->forward[i]->key < key)
{
node = node->forward[i];
}
update[i] = node;
}
node = node->forward[0];
if (node == nullptr || node->key != key)
{
int new_level = random_level();
if (new_level > level)
{
for (int i = level; i < new_level; i++)
{
update[i] = header;
}
level = new_level;
}
Node * new_node = new Node(key, new_level);
for (int i = 0; i < new_level; i++)
{
new_node->forward[i] = update[i]->forward[i];
update[i]->forward[i] = new_node;
}
return true;
}
return false;
}
bool SkipList::find(int key)
{
Node * node = header;
for (int i = level - 1; i >= 0; i--)
{
while (node->forward[i] && node->forward[i]->key <= key)
{
node = node->forward[i];
}
if (node->key == key)
return true;
}
return false;
}
bool SkipList::erase(int key)
{
Node * node = header;
Node * update[MAX_LEVEL];
fill(update, update + MAX_LEVEL, nullptr);
for (int i = level - 1; i >= 0; i--)
{
while (node->forward[i] && node->forward[i]->key < key)
{
node = node->forward[i];
}
update[i] = node;
}
node = node->forward[0];
if (node && node->key == key)
{
for (int i = 0; i < level; i++)
{
if (update[i]->forward[i] == node)
{
update[i]->forward[i] = node->forward[i];
}
}
delete node;
for (int i = level - 1; i >= 0; i--)
{
if (header->forward[i] == nullptr)
level--;
else
break;
}
}
return false;
}
void SkipList::print()
{
Node * node = nullptr;
for (int i = 0; i < level; i++)
{
node = header->forward[i];
cout << "Level " << i << " : ";
while (node)
{
cout << node->key << " ";
node = node->forward[i];
}
cout << endl;
}
cout << endl;
}
应用实例
在Lucene倒排列表中的应用
这里参考的是clucene-core-2.3.3.4的源码
可在下方链接获取源码:https://github.com/songqiyuan/CLucene
跳跃表存储以三层为例:
* Example for skipInterval = 3:
* c (skip level 2)
* c c c (skip level 1)
* x x x x x x x x x x (skip level 0)
* d d d d d d d d d d d d d d d d d d d d d d d d d d d d d d d d (posting list)
* 3 6 9 12 15 18 21 24 27 30 (df)
*
1.跳跃表可根据倒排表本身的长度(DocFreq)和跳跃的幅度(SkipInterval)而分不同的层次,层次数为NumSkipLevels = Min(MaxSkipLevels, floor(log(DocFreq/log(SkipInterval)))).
第Level层的节点数为DocFreq/(SkipInterval^(Level + 1)),level从零计数。
2.除最低层以外,其他层都有SkipChildLevelPointer来指向下一层相应的节点所在文件位置。
确定层数和跳跃表的构建代码如下:
/*SkipListWriter.cpp*/
void MultiLevelSkipListWriter::bufferSkip(int32_t df){
int32_t numLevels;
// determine max level
for (numLevels = 0; (df % skipInterval) == 0 && numLevels < numberOfSkipLevels; df /= skipInterval) {
numLevels++;
}
int64_t childPointer = 0;
for (int32_t level = 0; level < numLevels; level++) {
writeSkipData(level, (*skipBuffer)[level]);
int64_t newChildPointer = (*skipBuffer)[level]->getFilePointer();
if (level != 0) {
// store child pointers for all levels except the lowest
(*skipBuffer)[level]->writeVLong(childPointer);
}
//remember the childPointer for the next level
childPointer = newChildPointer;
}
}
MultiLevelSkipListWriter::MultiLevelSkipListWriter(int32_t skipInterval, int32_t maxSkipLevels, int32_t df) {
this->skipBuffer = NULL;
this->skipInterval = skipInterval;
// calculate the maximum number of skip levels for this document frequency
numberOfSkipLevels = df == 0 ? 0 : (int32_t) floor(log((float_t)df) / log((float_t)skipInterval));
// make sure it does not exceed maxSkipLevels
if (numberOfSkipLevels > maxSkipLevels) {
numberOfSkipLevels = maxSkipLevels;
}
}
关于数据存储,有机会再分析CLucene文件格式时,会详细介绍,现在了解即可,这里主要说,跳跃表在lucene中的应用。
3.除了最高层之外,其他层都有SkipLevelLength来表示此层的二进制长度(而非节点的个数),方便读取某一层的跳跃表到缓存里面。
4.低层在前,高层在后,当读完所有的低层后,剩下的就是最后一层,因而最后一层不需要SkipLevelLength。这也是为什么Lucene文档中的格式描述为 NumSkipLevels-1, SkipLevel,也即低NumSKipLevels-1层有SkipLevelLength,最后一层只有SkipLevel,没有SkipLevelLength。
5.每一个跳跃节点包含以下信息:文档号,payload的长度,文档号对应的倒排表中的节点在frq中的偏移量,文档号对应的倒排表中的节点在prx中的偏移量
在Redis中的应用
Redis使用跳跃表作为有序集合键的底层实现之一,如果一个有序集合包含的元素数量比较多,又或者有序集合中元素的成员(member)是比较长的字符串时,Redis就会使用跳跃表来作为有序集合键的底层实现
当有序集合对象可以同时满足以下两个条件时,对象使用ziplist(压缩链表)编码:
❑有序集合保存的元素数量小于128个;
❑有序集合保存的所有元素成员的长度都小于64字节;不能满足以上两个条件的有序集合对象将使用skiplist(跳表)编码。
注意以上两个条件的上限值是可以修改的,具体请看配置文件中关于zset-max-ziplist-entries选项和zset-max-ziplist-value选项的说明。
源码位置:http://download.redis.io/releases/
/*跳跃表的结构定义在server.h/zskiplist*/
/*3.0版本的定义在redis.h文件中*/
/*节点结构定义*/
/* ZSETs use a specialized version of Skiplists */
typedef struct zskiplistNode {
sds ele;
double score;
struct zskiplistNode *backward;
struct zskiplistLevel {
struct zskiplistNode *forward;
unsigned long span;
} level[];
} zskiplistNode;
typedef struct zskiplist {
struct zskiplistNode *header, *tail;
unsigned long length;
int level;
} zskiplist;
zskiplistj结构:
❑header:指向跳跃表的表头节点。
❑tail:指向跳跃表的表尾节点。
❑level:记录目前跳跃表内,层数最大的那个节点的层数(表头节点的层数不计算在内)。
❑length:记录跳跃表的长度,也即是,跳跃表目前包含节点的数量(表头节点不计算在内)。
zskiplistNode结构:
❑层(level):节点中用L1、L2、L3等字样标记节点的各个层,L1代表第一层,L2代表第二层,以此类推。每个层都带有两个属性:前进指针和跨度。前进指针用于访问位于表尾方向的其他节点,而跨度则记录了前进指针所指向节点和当前节点的距离。在上面的图片中,连线上带有数字的箭头就代表前进指针,而那个数字就是跨度。当程序从表头向表尾进行遍历时,访问会沿着层的前进指针进行。
❑后退(backward)指针:节点中用BW字样标记节点的后退指针,它指向位于当前节点的前一个节点。后退指针在程序从表尾向表头遍历时使用。
❑分值(score):各个节点中的1.0、2.0和3.0是节点所保存的分值。在跳跃表中,节点按各自所保存的分值从小到大排列。
❑成员对象(ele):各个节点中的o1、o2和o3是节点所保存的成员对象。
redis跳表创建、插入以及删除源码
这里先讲一下节点中跨度的含义即zskiplistNode 中变量( unsigned long span):
层的跨度(level[i].span属性)用于记录两个节点之间的距离:
❑两个节点之间的跨度越大,它们相距得就越远。
❑指向NULL的所有前进指针的跨度都为0,因为它们没有连向任何节点。
初看上去,很容易以为跨度和遍历操作有关,但实际上并不是这样,遍历操作只使用前进指针就可以完成了,跨度实际上是用来计算排位(rank)的:在查找某个节点的过程中,将沿途访问过的所有层的跨度累计起来,得到的结果就是目标节点在跳跃表中的排位。
有了前面的跳表基本知识,代码很容易理解、这里就不在赘述.
跳表中的排序是以分值的大小来排序的。具体代码
/*在redis中定义的最大层数和计算当前节点应为几层的概率*/
#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^64 elements */
#define ZSKIPLIST_P 0.25 /* Skiplist P = 1/4 */
/*源码位置:t_zset.c*/
zskiplistNode *zslInsert(zskiplist *zsl, double score, sds ele) {
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--) {
/* store rank that is crossed to reach the insert position */
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 &&
sdscmp(x->level[i].forward->ele,ele) < 0)))
{
rank[i] += x->level[i].span;
x = x->level[i].forward;
}
update[i] = x;
}
/* we assume the element is not already inside, since we allow duplicated
* scores, reinserting the same element should never happen since the
* caller of zslInsert() should test in the hash table if the element is
* already inside or not. */
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,ele);
for (i = 0; i < level; i++) {
x->level[i].forward = update[i]->level[i].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]);
update[i]->level[i].span = (rank[0] - rank[i]) + 1;
}
/* increment span for untouched levels */
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;
}
/* Internal function used by zslDelete, zslDeleteRangeByScore and
* zslDeleteRangeByRank. */
void zslDeleteNode(zskiplist *zsl, zskiplistNode *x, zskiplistNode **update) {
int i;
for (i = 0; i < zsl->level; i++) {
if (update[i]->level[i].forward == x) {
update[i]->level[i].span += x->level[i].span - 1;
update[i]->level[i].forward = x->level[i].forward;
} else {
update[i]->level[i].span -= 1;
}
}
if (x->level[0].forward) {
x->level[0].forward->backward = x->backward;
} else {
zsl->tail = x->backward;
}
while(zsl->level > 1 && zsl->header->level[zsl->level-1].forward == NULL)
zsl->level--;
zsl->length--;
}
/* Delete an element with matching score/element from the skiplist.
* The function returns 1 if the node was found and deleted, otherwise
* 0 is returned.
*
* If 'node' is NULL the deleted node is freed by zslFreeNode(), otherwise
* it is not freed (but just unlinked) and *node is set to the node pointer,
* so that it is possible for the caller to reuse the node (including the
* referenced SDS string at node->ele). */
int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
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 &&
sdscmp(x->level[i].forward->ele,ele) < 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->level[0].forward;
if (x && score == x->score && sdscmp(x->ele,ele) == 0) {
zslDeleteNode(zsl, x, update);
if (!node)
zslFreeNode(x);
else
*node = x;
return 1;
}
return 0; /* not found */
}