skiplist(跳表)的原理及JAVA实现 前记 一.跳表的原理及思想 二. 自己动手用JAVA实现SkipList跳表 三. 分析JDK实现的跳表ConcurrentSkipListMap 四. 跳表的应用场景
前记
最近在看Redis,之间就尝试用sortedSet用在实现排行榜的项目，那么sortedSet底层是什么结构呢？ "Redis sorted set的内部使用HashMap和跳跃表(SkipList)来保证数据的存储和有序，HashMap里放的是成员到score的映射，而跳跃表里存放的是所有的成员，排序依据是HashMap里存的score,使用跳跃表的结构可以获得比较高的查找效率，并且在实现上比较简单。” 那么什么是SkipList跳表呢？下面我们从理解它的思想到实现及应用去做一个大致的了解。
一.跳表的原理及思想
跳表的背景
Skip list是一个用于有序元素序列快速搜索的数据结构，由美国计算机科学家William Pugh发明于1989年。他在论文《Skip lists: a probabilistic alternative to balanced trees》中详细介绍了跳表的数据结构和插入删除等操作。论文是这么介绍跳表的：
Skip lists are a data structure that can be used in place of balanced trees.
Skip lists use probabilistic balancing rather than strictly enforced balancing and as a result the algorithms for insertion and deletion in skip lists are much simpler and significantly faster than equivalent algorithms for balanced trees.
也就是说，
Skip list是一个“概率型”的数据结构，可以在很多应用场景中替代平衡树。Skip list算法与平衡树相比，有相似的渐进期望时间边界，但是它更简单，更快，使用更少的空间。
Skip list是一个分层结构多级链表，最下层是原始的链表，每个层级都是下一个层级的“高速跑道”。
为什么选择跳表
目前经常使用的平衡数据结构有：B树，红黑树，AVL树，Splay Tree, Treep等。
想象一下，给你一张草稿纸，一只笔，一个编辑器，你能立即实现一颗红黑树，或者AVL树
出来吗？ 很难吧，这需要时间，要考虑很多细节，要参考一堆算法与数据结构之类的树，
还要参考网上的代码，相当麻烦。
用跳表吧，跳表是一种随机化的数据结构，目前开源软件 Redis 和 LevelDB 都有用到它，
它的效率和红黑树以及 AVL 树不相上下，但跳表的原理相当简单，只要你能熟练操作链表，就能轻松实现一个 SkipList。
有序表的搜索
考虑一个有序表：
从该有序表中搜索元素 < 23, 43, 59 > ，需要比较的次数分别为 < 2, 4, 6 >，总共比较的次数
为 2 + 4 + 6 = 12 次。有没有优化的算法吗? 链表是有序的，但不能使用二分查找。类似二叉
搜索树，我们把一些节点提取出来，作为索引。得到如下结构：
这里我们把 < 14, 34, 50, 72 > 提取出来作为一级索引，这样搜索的时候就可以减少比较次数了。
我们还可以再从一级索引提取一些元素出来，作为二级索引，变成如下结构：
这里元素不多，体现不出优势，如果元素足够多，这种索引结构就能体现出优势来了。
这基本上就是跳表的核心思想，其实也是一种通过“空间来换取时间”的一个算法，通过在每个节点中增加了向前的指针，从而提升查找的效率。
跳表
下面的结构是就是跳表：
其中 -1 表示 INT_MIN， 链表的最小值，1 表示 INT_MAX，链表的最大值。
跳表具有如下性质：
(1) 由很多层结构组成
(2) 每一层都是一个有序的链表
(3) 最底层(Level 1)的链表包含所有元素
(4) 如果一个元素出现在 Level i 的链表中，则它在 Level i 之下的链表也都会出现。
(5) 每个节点包含两个指针，一个指向同一链表中的下一个元素，一个指向下面一层的元素。
跳表的搜索
例子：查找元素 117
(1) 比较 21， 比 21 大，往后面找
(2) 比较 37, 比 37大，比链表最大值小，从 37 的下面一层开始找
(3) 比较 71, 比 71 大，比链表最大值小，从 71 的下面一层开始找
(4) 比较 85， 比 85 大，从后面找
(5) 比较 117， 等于 117， 找到了节点。
二. 自己动手用JAVA实现SkipList跳表
单纯的用链表来实现一个SkipList。
基本Node结构
package com.shoshana.skiplist;
public class SkipListNode<T> {
public int key;
public T value;
public SkipListNode<T> pre, next, up, down; //上下左右四个节点，pre和up存在的意义在于 "升层"的时候需要查找相邻节点
public static final int HEAD_KEY = Integer.MIN_VALUE; // 负无穷
public static final int TAIL_KEY = Integer.MAX_VALUE; // 正无穷
public SkipListNode(int k, T v) {
key = k;
value = v;
}
public int getKey() {
return key;
}
public void setKey(int key) {
this.key = key;
}
public T getValue() {
return value;
}
public void setValue(T value) {
this.value = value;
}
public boolean equals(Object o) {
if (this == o) {
return true;
}
if (o == null) {
return false;
}
if (!(o instanceof SkipListNode<?>)) {
return false;
}
SkipListNode<T> ent;
try {
ent = (SkipListNode<T>) o; //检测类型
} catch (ClassCastException ex) {
return false;
}
return (ent.getKey() == key) && (ent.getValue() == value);
}
@Override
public String toString() {
return "key-value:" + key + "," + value;
}
}
跳表实现
package com.shoshana.skiplist;
import java.util.Random;
public class SkipList<T> {
private SkipListNode<T> head, tail;
private int size;
private int listLevel;
private Random random;
private static final double PROBABILITY = 0.5;
public SkipList() {
head = new SkipListNode<T>(SkipListNode.HEAD_KEY, null);
tail = new SkipListNode<>(SkipListNode.TAIL_KEY, null);
head.next = tail;
tail.pre = head;
size = 0;
listLevel = 0;
random = new Random();
}
public SkipListNode<T> get(int key) {
SkipListNode<T> p = findNode(key);
if (p.key == key) {
return p;
}
return null;
}
//首先查找到包含key值的节点，将节点从链表中移除，接着如果有更高level的节点，则repeat这个操作即可。
public T remove(int k) {
SkipListNode<T> p = get(k);
if (p == null) {
return null;
}
T oldV = p.value;
SkipListNode<T> q;
while (p != null) {
q = p.next;
q.pre = p.pre;
p.pre.next = q;
p = p.up;
}
return oldV;
}
/**
* put方法有一些需要注意的步骤：
* 1.如果put的key值在跳跃表中存在，则进行修改操作；
* 2.如果put的key值在跳跃表中不存在，则需要进行新增节点的操作，并且需要由random随机数决定新加入的节点的高度（最大level）；
* 3.当新添加的节点高度达到跳跃表的最大level，需要添加一个空白层（除了-oo和+oo没有别的节点）
*
* @param k
* @param v
*/
public void put(int k, T v) {
System.out.println("添加key:" + k);
SkipListNode<T> p = findNode(k);//这里不用get是因为下面可能用到这个节点
System.out.println("找到P:" + p);
if (p.key == k) {
p.value = v;
return;
}
SkipListNode<T> q = new SkipListNode<>(k, v);
insertNode(p, q);
int currentLevel = 0;
while (random.nextDouble() > PROBABILITY) {
if (currentLevel >= listLevel) {
addEmptyLevel();
System.out.println("升层");
}
while (p.up == null) {
System.out.println(p);
p = p.pre;
System.out.println("找到第一个有上层结点的值" + p);
}
p = p.up;
//创建 q的镜像变量（只存储k，不存储v，因为查找的时候会自动找最底层数据）
SkipListNode<T> z = new SkipListNode<>(k, null);
insertNode(p, z);
z.down = q;
q.up = z;
//别忘了把指针移到上一层。
q = z;
currentLevel++;
System.out.println("添加后" + this);
}
size++;
}
/**
* 如果传入的key值在跳跃表中不存在，则findNode返回跳跃表中key值小于key，并且key值相差最小的底层节点;
* 所以不能用此方法来代替get
*
* @param key
* @return
*/
public SkipListNode<T> findNode(int key) {
SkipListNode<T> p = head;
while (true) {
System.out.println("p.next.key:" + p.next.key);
if (p.next != null && p.next.key <= key) {
p = p.next;
}
System.out.println("找到node:" + p);
if (p.down != null) {
System.out.println("node.down :" + p);
p = p.down;
} else if (p.next != null && p.next.key > key) {
break;
}
}
return p;
}
public boolean isEmpty() {
return size == 0;
}
public int size() {
return size;
}
public void addEmptyLevel() {
SkipListNode<T> p1 = new SkipListNode<T>(SkipListNode.HEAD_KEY, null);
SkipListNode<T> p2 = new SkipListNode<T>(SkipListNode.TAIL_KEY, null);
p1.next = p2;
p1.down = head;
p2.pre = p1;
p2.down = tail;
head.up = p1;
tail.up = p2;
head = p1;
tail = p2;
listLevel++;
}
private void insertNode(SkipListNode<T> p, SkipListNode<T> q) {
q.next = p.next;
q.pre = p;
p.next.pre = q;
p.next = q;
}
public int getLevel() {
return listLevel;
}
}
Demo及运行
package com.shoshana.skiplist;
public class SkipListDemo {
public static void main(String[] args) {
SkipList<String> list = new SkipList<String>();
list.put(10, "sho");
list.put(1, "sha");
list.put(9, "na");
list.put(2, "bing");
list.put(8, "ling");
list.put(7, "xiao");
list.put(100, "你好，skiplist");
list.put(5, "冰");
list.put(6, "灵");
System.out.println("列表元素：
" + list);
System.out.println("删除100：" + list.remove(100));
System.out.println("列表元素：
" + list);
System.out.println("5对于的value：
" + list.get(5).value);
System.out.println("链表大小：" + list.size() + ",深度：" + list.getLevel());
}
}
运行结果：
classpath "C:Program com.shoshana.skiplist.SkipListDemo 添加key:10 p.next.key:2147483647 找到node:key-value:-2147483648,null 找到P:key-value:-2147483648,null 升层 添加后com.shoshana.skiplist.SkipList@74a14482 添加key:1 p.next.key:10 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:10 找到node:key-value:-2147483648,null 找到P:key-value:-2147483648,null 添加key:9 p.next.key:10 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:1 找到node:key-value:1,sha 找到P:key-value:1,sha 添加key:2 p.next.key:10 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:1 找到node:key-value:1,sha 找到P:key-value:1,sha key-value:1,sha 找到第一个有上层结点的值key-value:-2147483648,null 添加后com.shoshana.skiplist.SkipList@74a14482 添加key:8 p.next.key:2 找到node:key-value:2,null node.down :key-value:2,null p.next.key:9 找到node:key-value:2,bing 找到P:key-value:2,bing 添加key:7 p.next.key:2 找到node:key-value:2,null node.down :key-value:2,null p.next.key:8 找到node:key-value:2,bing 找到P:key-value:2,bing 添加后com.shoshana.skiplist.SkipList@74a14482 升层 key-value:2,null 找到第一个有上层结点的值key-value:-2147483648,null 添加后com.shoshana.skiplist.SkipList@74a14482 升层 添加后com.shoshana.skiplist.SkipList@74a14482 添加key:100 p.next.key:7 找到node:key-value:7,null node.down :key-value:7,null p.next.key:2147483647 找到node:key-value:7,null node.down :key-value:7,null p.next.key:10 找到node:key-value:10,null node.down :key-value:10,null p.next.key:2147483647 找到node:key-value:10,sho 找到P:key-value:10,sho 添加后com.shoshana.skiplist.SkipList@74a14482 key-value:10,null 找到第一个有上层结点的值key-value:7,null 添加后com.shoshana.skiplist.SkipList@74a14482 添加key:5 p.next.key:7 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:7 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:2 找到node:key-value:2,null node.down :key-value:2,null p.next.key:7 找到node:key-value:2,bing 找到P:key-value:2,bing 添加key:6 p.next.key:7 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:7 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:2 找到node:key-value:2,null node.down :key-value:2,null p.next.key:5 找到node:key-value:5,冰 找到P:key-value:5,冰 key-value:5,冰 找到第一个有上层结点的值key-value:2,bing 添加后com.shoshana.skiplist.SkipList@74a14482 key-value:2,null 找到第一个有上层结点的值key-value:-2147483648,null 添加后com.shoshana.skiplist.SkipList@74a14482 添加后com.shoshana.skiplist.SkipList@74a14482 列表元素： com.shoshana.skiplist.SkipList@74a14482 p.next.key:6 找到node:key-value:6,null node.down :key-value:6,null p.next.key:7 找到node:key-value:7,null node.down :key-value:7,null p.next.key:10 找到node:key-value:10,null node.down :key-value:10,null p.next.key:100 找到node:key-value:100,你好，skiplist 删除100：你好，skiplist 列表元素： com.shoshana.skiplist.SkipList@74a14482 p.next.key:6 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:6 找到node:key-value:-2147483648,null node.down :key-value:-2147483648,null p.next.key:2 找到node:key-value:2,null node.down :key-value:2,null p.next.key:5 找到node:key-value:5,冰 5对于的value： 冰 链表大小：9,深度：3 Process finished with exit code 0
三. 分析JDK实现的跳表ConcurrentSkipListMap
在JDK内部，也使用了该数据结构，比如ConcurrentSkipListMap,ConcurrentSkipListSet等。下面我们主要介绍ConcurrentSkipListMap。说到ConcurrentSkipListMap，我们就应该比较HashMap，ConcurrentHashMap，ConcurrentSkipListMap这三个类来讲解。它们都是以键值对的方式来存储数据的。HashMap是线程不安全的，而ConcurrentHashMap和ConcurrentSkipListMap是线程安全的，它们内部都使用无锁CAS算法实现了同步。ConcurrentHashMap中的元素是无序的，ConcurrentSkipListMap中的元素是有序的。它们三者的具体区别可以参考具体的资料，下面主要讲解ConcurrentSkipListMap的实现原理。
ConcurrentSkipListMap提供了一种线程安全的并发访问的排序映射表。内部是SkipList（跳表）结构实现，在理论上能够在O(log(n))时间内完成查找、插入、删除操作。注意，调用ConcurrentSkipListMap的size时，由于多个线程可以同时对映射表进行操作，所以映射表需要遍历整个链表才能返回元素个数，这个操作是个O(log(n))的操作。
doPut()
private V doPut(K kkey, V value, boolean onlyIfAbsent) {
Comparable<? super K> key = comparable(kkey);
for (;;) {
// 找到key的前继节点
Node<K,V> b = findPredecessor(key);
// 设置n为“key的前继节点的后继节点”，即n应该是“插入节点”的“后继节点”
Node<K,V> n = b.next;
for (;;) {
if (n != null) {
Node<K,V> f = n.next;
// 如果两次获得的b.next不是相同的Node，就跳转到”外层for循环“，重新获得b和n后再遍历。
if (n != b.next)
break;
// v是“n的值”
Object v = n.value;
// 当n的值为null(意味着其它线程删除了n)；此时删除b的下一个节点，然后跳转到”外层for循环“，重新获得b和n后再遍历。
if (v == null) { // n is deleted
n.helpDelete(b, f);
break;
}
// 如果其它线程删除了b；则跳转到”外层for循环“，重新获得b和n后再遍历。
if (v == n || b.value == null) // b is deleted
break;
// 比较key和n.key
int c = key.compareTo(n.key);
if (c > 0) {
b = n;
n = f;
continue;
}
if (c == 0) {
if (onlyIfAbsent || n.casValue(v, value))
return (V)v;
else
break; // restart if lost race to replace value
}
// else c < 0; fall through
}
// 新建节点(对应是“要插入的键值对”)
Node<K,V> z = new Node<K,V>(kkey, value, n);
// 设置“b的后继节点”为z
if (!b.casNext(n, z))
break; // 多线程情况下，break才可能发生(其它线程对b进行了操作)
// 随机获取一个level
// 然后在“第1层”到“第level层”的链表中都插入新建节点
int level = randomLevel();
if (level > 0)
insertIndex(z, level);
return null;
}
}
}
doRemove
final V doRemove(Object okey, Object value) {
Comparable<? super K> key = comparable(okey);
for (;;) {
// 找到“key的前继节点”
Node<K,V> b = findPredecessor(key);
// 设置n为“b的后继节点”(即若key存在于“跳表中”，n就是key对应的节点)
Node<K,V> n = b.next;
for (;;) {
if (n == null)
return null;
// f是“当前节点n的后继节点”
Node<K,V> f = n.next;
// 如果两次读取到的“b的后继节点”不同(其它线程操作了该跳表)，则返回到“外层for循环”重新遍历。
if (n != b.next) // inconsistent read
break;
// 如果“当前节点n的值”变为null(其它线程操作了该跳表)，则返回到“外层for循环”重新遍历。
Object v = n.value;
if (v == null) { // n is deleted
n.helpDelete(b, f);
break;
}
// 如果“前继节点b”被删除(其它线程操作了该跳表)，则返回到“外层for循环”重新遍历。
if (v == n || b.value == null) // b is deleted
break;
int c = key.compareTo(n.key);
if (c < 0)
return null;
if (c > 0) {
b = n;
n = f;
continue;
}
// 以下是c=0的情况
if (value != null && !value.equals(v))
return null;
// 设置“当前节点n”的值为null
if (!n.casValue(v, null))
break;
// 设置“b的后继节点”为f
if (!n.appendMarker(f) || !b.casNext(n, f))
findNode(key); // Retry via findNode
else {
// 清除“跳表”中每一层的key节点
findPredecessor(key); // Clean index
// 如果“表头的右索引为空”，则将“跳表的层次”-1。
if (head.right == null)
tryReduceLevel();
}
return (V)v;
}
}
}
findNode
private Node<K,V> findNode(Comparable<? super K> key) {
for (;;) {
// 找到key的前继节点
Node<K,V> b = findPredecessor(key);
// 设置n为“b的后继节点”(即若key存在于“跳表中”，n就是key对应的节点)
Node<K,V> n = b.next;
for (;;) {
// 如果“n为null”，则跳转中不存在key对应的节点，直接返回null。
if (n == null)
return null;
Node<K,V> f = n.next;
// 如果两次读取到的“b的后继节点”不同(其它线程操作了该跳表)，则返回到“外层for循环”重新遍历。
if (n != b.next) // inconsistent read
break;
Object v = n.value;
// 如果“当前节点n的值”变为null(其它线程操作了该跳表)，则返回到“外层for循环”重新遍历。
if (v == null) { // n is deleted
n.helpDelete(b, f);
break;
}
if (v == n || b.value == null) // b is deleted
break;
// 若n是当前节点，则返回n。
int c = key.compareTo(n.key);
if (c == 0)
return n;
// 若“节点n的key”小于“key”，则说明跳表中不存在key对应的节点，返回null
if (c < 0)
return null;
// 若“节点n的key”大于“key”，则更新b和n，继续查找。
b = n;
n = f;
}
}
}
四. 跳表的应用场景
Java API中提供了支持并发操作的跳跃表ConcurrentSkipListSet和ConcurrentSkipListMap。
有序的情况下：
在非多线程的情况下，应当尽量使用TreeMap（红黑树实现）。
对于并发性相对较低的并行程序可以使用Collections.synchronizedSortedMap将TreeMap进行包装，也可以提供较好的效率。
但是对于高并发程序，应当使用ConcurrentSkipListMap。
无序情况下：
并发程度低，数据量大时，ConcurrentHashMap 存取远大于ConcurrentSkipListMap。
数据量一定，并发程度高时，ConcurrentSkipListMap比ConcurrentHashMap效率更高。