简单高效的跳跃表 ConcurrentSkipListMap
一、简介
看了TreeMap、HashMap源码的，或者手写实现红黑树、平衡二叉树的，感受插入和删除手写太难了；我逻辑理解也不是很顺畅，插入基本搞清楚了，可是删除，我就呵呵了node
那么，有没有一种性能和红黑树或者平衡二叉树不相上下，且又很好实现的数据结构，答案是确定，我知道的至少有一种 跳表；不过也有缺点，看实现方式（我找的资料都是插入时，对刚插入的数据随机处理索引），不必定稳定安全
什么叫跳表呢？ 跳表是一个随机化的有序链表。跳表在原有的有序链表上面增长了多级索引，经过索引来实现快速查找。跳表不只能提升搜索性能，同时也能够提升插入和删除操做的性能。bash
它采用随机技术决定链表中哪些节点应增长向前指针以及在该节点中应增长多少个指针。 采用这种随机技术，跳表中的搜索、插入、删除操做的时间均为O(logn)，然而，最坏状况下时间复杂性却变成O(n)。 跳表的结构大体以下图：（来源百度百科）数据结构
二、ConcurrentSkipListMap源码分析
2.1 数据节点
final K key;
volatile Object value;
volatile Node<K,V> next;
复制代码
节点数据包含，key，value和下个数据；也就是采用单链表的结构存储实际数据app
具体的代码操做就不贴出了，大概有如下内容：dom
- 使用sun.misc.Unsafe类来实现原子操做;配合volatile关键+自旋能够实现线程安全操做
- 实现了值改变原子操做，后驱节点变化原子操做，当前next节点删除的线程安全操做方法
2.2 普通索引节点
final Node<K,V> node;
final Index<K,V> down;
volatile Index<K,V> right;
复制代码
包含对应数据，下方索引节点，右边索引节点；源码分析
- 一个特殊的单链表，和下方索引节点的数据节点相同；
- 实现右边索引节点的原子操做
2.3 头索引节点 HeadIndex<K,V>
static final class HeadIndex<K,V> extends Index<K,V> {
final int level;
HeadIndex(Node<K,V> node, Index<K,V> down, Index<K,V> right, int level) {
super(node, down, right);
this.level = level;
}
}
复制代码
继承了普通索引节点，增长了索引级别，从1开始；post
2.4 主要成员变量
static final Object BASE_HEADER = new Object();
private transient volatile HeadIndex<K,V> head;
final Comparator<? super K> comparator;
复制代码
head 索引节点，comparator比较器；特色性能
- 头索引节点，存储的值对为 null-BASE_HEADER;数据节点第一个值为BASE_HEADER
- 头索引节点的变化实现了原子操做
- head值为最高索引级别（level最大）的第一个索引节点
- 存储key值实现了Comparator接口，或者提供了比较器comparator，否则比较时抛出异常
2.5 查找前驱节点
private Node<K,V> findPredecessor(Object key, Comparator<? super K> cmp) {
if (key == null)
throw new NullPointerException(); // don't postpone errors for (;;) { for (Index<K,V> q = head, r = q.right, d;;) { if (r != null) { Node<K,V> n = r.node; K k = n.key; if (n.value == null) { if (!q.unlink(r)) break; r = q.right; continue; } if (cpr(cmp, key, k) > 0) { q = r; r = r.right; continue; } } if ((d = q.down) == null) return q.node; q = d; r = d.right; } } } 复制代码
- 首先判断key值，为空，抛出异常
- 采用双层for循环；在原子性操做去除无效值失败时，内循环从新开始，至关于自旋操做，保证线程同步
- 内层循环采用right索引的值域比较大小：
- right索引不为空，首先若是rigth索引的值域无效（值节点中value为空），则原子操做（unlink方法）连接q节点和right索引的rigt索引，失败从新循环，成功，索引向右移动
- right索引不为空，若是right索引的值-key 比当前key小（cpr(cmp, key, k)， key大，返回 > 0），则继续往右移动，进行下次循环
- right索引为空，或者小，则索引向下移动
- 若是down指针域为空，则说明，这就是在当前全部索引节点中，小于key的最大索引节点
向下移动，则是把down指针域当成next，进行链表移动ui
向右移动，则是把right指针域当成next，进行链表移动
2.6 查找节点 Node<K,V> findNode(Object key)
private Node<K,V> findNode(Object key) {
if (key == null)
throw new NullPointerException();
Comparator<? super K> cmp = comparator;
outer: for (;;) {
for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
Object v; int c;
if (n == null)
break outer;
Node<K,V> f = n.next;
if (n != b.next)
break;
if ((v = n.value) == null) {
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n)
break;
if ((c = cpr(cmp, key, n.key)) == 0)
return n;
if (c < 0)
break outer;
b = n;
n = f;
}
}
return null;
}
复制代码
- key值检验有效性
- 双层循环；内循环处理流程
- 若是next节点为空，则说明未找到匹配的数据，则跳出双层循环，结束流程，返回null
- 检查数据是否发生了变化，若是发生变化，从新内循环
- 检查数据有效性，无效时，去除节点（cas操做，有可能失败），从新循环
- 检查b、n数据有效性，无效从新内循环
- 比较索引值域-key 和key的大小，至关代表找到数据，返回；小于0，则不存在，跳出双层循环，返回null；大于0，继续日后找
因而可知，数据节点按照key的从小到大排列的
2.7 获取数据 get
public V get(Object key) {
return doGet(key);
}
private V doGet(Object key) {
if (key == null)
throw new NullPointerException();
Comparator<? super K> cmp = comparator;
outer: for (;;) {
for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
Object v; int c;
if (n == null)
break outer;
Node<K,V> f = n.next;
if (n != b.next)
break;
if ((v = n.value) == null) {
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n)
break;
if ((c = cpr(cmp, key, n.key)) == 0) {
@SuppressWarnings("unchecked") V vv = (V)v;
return vv;
}
if (c < 0)
break outer;
b = n;
n = f;
}
}
return null;
}
复制代码
和findNode方法基本一致，就是在下面代码中，返回值不一样；一个是返回节点，一个是返回节点value值
if ((c = cpr(cmp, key, n.key)) == 0) {
@SuppressWarnings("unchecked") V vv = (V)v;
return vv;
}
复制代码
2.8 增长、修改数据
public V put(K key, V value) {
if (value == null)
throw new NullPointerException();
return doPut(key, value, false);
}
private V doPut(K key, V value, boolean onlyIfAbsent) {
Node<K,V> z;
if (key == null)
throw new NullPointerException();
Comparator<? super K> cmp = comparator;
outer: for (;;) {
for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
if (n != null) {
Object v; int c;
Node<K,V> f = n.next;
if (n != b.next)
break;
if ((v = n.value) == null) {
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n)
break;
if ((c = cpr(cmp, key, n.key)) > 0) {
b = n;
n = f;
continue;
}
if (c == 0) {
if (onlyIfAbsent || n.casValue(v, value)) {
@SuppressWarnings("unchecked") V vv = (V)v;
return vv;
}
break; // restart if lost race to replace value
}
}
z = new Node<K,V>(key, value, n);
if (!b.casNext(n, z))
break;
break outer;
}
}
int rnd = ThreadLocalRandom.nextSecondarySeed();
if ((rnd & 0x80000001) == 0) {
int level = 1, max;
while (((rnd >>>= 1) & 1) != 0)
++level;
Index<K,V> idx = null;
HeadIndex<K,V> h = head;
if (level <= (max = h.level)) {
for (int i = 1; i <= level; ++i)
idx = new Index<K,V>(z, idx, null);
}
else {
level = max + 1;
@SuppressWarnings("unchecked")Index<K,V>[] idxs =
(Index<K,V>[])new Index<?,?>[level+1];
for (int i = 1; i <= level; ++i)
idxs[i] = idx = new Index<K,V>(z, idx, null);
for (;;) {
h = head;
int oldLevel = h.level;
if (level <= oldLevel) // lost race to add level
break;
HeadIndex<K,V> newh = h;
Node<K,V> oldbase = h.node;
for (int j = oldLevel+1; j <= level; ++j)
newh = new HeadIndex<K,V>(oldbase, newh, idxs[j], j);
if (casHead(h, newh)) {
h = newh;
idx = idxs[level = oldLevel];
break;
}
}
}
splice: for (int insertionLevel = level;;) {
int j = h.level;
for (Index<K,V> q = h, r = q.right, t = idx;;) {
if (q == null || t == null)
break splice;
if (r != null) {
Node<K,V> n = r.node;
int c = cpr(cmp, key, n.key);
if (n.value == null) {
if (!q.unlink(r))
break;
r = q.right;
continue;
}
if (c > 0) {
q = r;
r = r.right;
continue;
}
}
if (j == insertionLevel) {
if (!q.link(r, t))
break;
if (t.node.value == null) {
findNode(key);
break splice;
}
if (--insertionLevel == 0)
break splice;
}
if (--j >= insertionLevel && j < level)
t = t.down;
q = q.down;
r = q.right;
}
}
}
return null;
}
复制代码
从上面代码能够看出，分两个大步骤：一是插入数据域节点（修改值只有这一步），二是索引层处理
插入数据：和get思路大同小异，存在一下区别
- 找到相同key的数据节点时，替换value值，结束整个流程
- 若是向后查找，当前节点next大于要插入key值，说明插入节点在当前节点和后面节点之间，原子操做连接，成功进行下一步
索引层处理
- 随机一个正数，若是不知足(rnd & 0x80000001) == 0，则结束
- 经过随机数rnd，来肯定作多有多少层level
- 若是level层数小于现有层，则直接生成当前数据的从1到level的索引节点
- 若是level层大于现有层，则增长一层：生成当前数据从1到level的索引节点，并生成原有层到新的level层的头节点，头节点右值域为当前数据同层索引节点；h为新的head指针，idxs为当前插入索引节点在旧的索引层最高层
- 双层循环：外层循环还是检验数据发生变化时从新处理，保证数据的线程安全
- 若是当前插入数据表明的索引节点为空，或者是，上次循环时索引层为1，结束
- 去除无效索引节点；成功，向右移动，继续循环，失败则从外层循环从新开始；
- 当前插入数据key值大，继续右移，继续循环
- 这时已经找到位置；找到位置的level和插入数据的level同层，则索引节点插入；插入为原子操做，失败，则从内层循环从新开始
- 找到位置的level和插入数据的level同层，存在当前插入索引为无效数据，则跳出外层循环，结束
- 找到位置的level和插入数据的level同层；插入数据，level 减1，若是此时值为0，说明索引节点所有连接成功，结束
- 若是当前内循环时索引节点level，比插入数据level大，则内循环索引节点向下移动，继续循环
- 若是内循环索引节点不大于插入数据level节点，则一块儿向下移动（在我看来，这就应该是同级移动），继续循环
2.9 移除数据 remove
public V remove(Object key) {
return doRemove(key, null);
}
final V doRemove(Object key, Object value) {
if (key == null)
throw new NullPointerException();
Comparator<? super K> cmp = comparator;
outer: for (;;) {
for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
Object v; int c;
if (n == null)
break outer;
Node<K,V> f = n.next;
if (n != b.next) // inconsistent read
break;
if ((v = n.value) == null) { // n is deleted
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n) // b is deleted
break;
if ((c = cpr(cmp, key, n.key)) < 0)
break outer;
if (c > 0) {
b = n;
n = f;
continue;
}
/////////////////////////////////////////////// 相等
if (value != null && !value.equals(v))
break outer;
if (!n.casValue(v, null))
break;
if (!n.appendMarker(f) || !b.casNext(n, f))
findNode(key); // retry via findNode
else {
findPredecessor(key, cmp); // clean index
if (head.right == null)
tryReduceLevel();
}
@SuppressWarnings("unchecked") V vv = (V)v;
return vv;
}
}
return null;
}
复制代码
又是熟悉的套路；校验，双层循环；那咱们从key值比较相等（代码中 /////////////////////////////////////////////// 相等 处）开始提及
- 若是删除数据的value值不为空且不相等，代表不存在想删除数据，跳出双层循环，结束
- 当前数据节点是想删除的数据节点，数据value原子操做置空，失败，内循环从新开始
- 删除数据value = null，原子操做去除数据节点并进行后驱处理，失败，则经过findNode来去除节点（其经过自旋+cas操做必定能操做成功）
- 删除数据节点，并处理后驱成功；则经过 findPredecessor方法进行无效索引节点去除，一样此操做确定能够成功；
- 若是最高层head索引节点，右指针域为空，尝试去除最高层索引层（已经没有任何数据域索引节点了）
- 返回删除节点的value值
2.10 释放索引层 tryReduceLevel
private void tryReduceLevel() {
HeadIndex<K,V> h = head;
HeadIndex<K,V> d;
HeadIndex<K,V> e;
if (h.level > 3 &&
(d = (HeadIndex<K,V>)h.down) != null &&
(e = (HeadIndex<K,V>)d.down) != null &&
e.right == null &&
d.right == null &&
h.right == null &&
casHead(h, d) &&
h.right != null)
casHead(d, h);
}
复制代码
去除索引条件
- 至少有4层索引节点
- head节点的down指针域不为空
- head节点的down指针域的down指针域不为空
- head的right指针域为空
- head的down指针域的右域为空
- head节点的down指针域的down指针域的右域为空
- 原子操做 head 置换为head的down指针域，失败，结束
- 当前head的右指针域不为空，则原子操做head，从新回退为h
为啥要这么多判断啊，为啥要回退啊，我是彻底不懂，只能说确定是为了处理线程同步问题的，哪位读者知道，谢谢留言告诉我
三、总结
- 存储包括数据单链表、索引单链表结构；索引单链表有两个指针域，右、下
- 数据单链表，是从小到大排列的
- 索引数据域，每一个数据对应的索引层数不是有序分布的
- 增长数据时，随机数决定是否增长索引层，是否增长插入值索引；这个随机很重要，影响着效率
- 删除数据后，可能减小索引层
- head索引以及其down、down-down指针域的节点值是固定的
- 未加入数据时有一层索引，一个固定头数据
- 查找时，经过索引查找最大左区间数据节点，而后再数据链表中向后查询
- 插入数据的key类，必须实现Comparator接口或者初始化时传入比较器
技术变化都很快，但基础技术、理论知识永远都是那些；做者但愿在余后的生活中，对经常使用技术点进行基础知识分享；若是你以为文章写的不错，请给与关注和点赞；若是文章存在错误，也请多多指教！
相关文章
- 1. 基于跳跃表的 ConcurrentSkipListMap 内部实现（Java 8）
- 2. 跳跃表
- 3. 跳跃表（跳表）
- 4. 跳~跳~跳跃表
- 5. ConcurrentSkipListMap跳表原理解析
- 6. redis的跳跃表
- 7. Redis跳跃表
- 8. Redis——跳跃表
- 9. 跳跃表-skiplist
- 10. redis跳跃表
- 更多相关文章...
- • jQuery Mobile 表单 - jQuery Mobile 教程
- • Thymeleaf表单 - Thymeleaf 教程
- • Github 简明教程
- • TiDB 在摩拜单车在线数据业务的应用和实践
相关标签/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
欢迎关注本站公众号,获取更多信息
相关文章
- 1. 基于跳跃表的 ConcurrentSkipListMap 内部实现（Java 8）
- 2. 跳跃表
- 3. 跳跃表（跳表）
- 4. 跳~跳~跳跃表
- 5. ConcurrentSkipListMap跳表原理解析
- 6. redis的跳跃表
- 7. Redis跳跃表
- 8. Redis——跳跃表
- 9. 跳跃表-skiplist
- 10. redis跳跃表