Java并发容器之ConcurrentSkipListMap源码分析
一、概述
在学习ConcurrentSkipListMap之前，我们需要来了解一种随机化的数据结构：跳跃表（skip list）。
1. 跳跃表
在我们通常所使用到的数据结构中，使用最多的无外乎就是数组和链表，而数组和链表各自的优缺点也很明显：
数组的优点在于元素的查找，链表的优点在于添加或移除等操作；而对一个有序的数组而言，添加一个元素，可能需要移动大量的元素；而对于一个有序的链表而言，查找一个元素，又不能通过像在数组查找中使用二分查找，只能通过顺序遍历的方式进行。
而为了使链表查询的时候也有一个不错的性能，所以就出现了跳跃表（skip list）这种数据结构。
2. 介绍
我们以一个链表查询的小例子来介绍跳跃表，顺便介绍下跳跃表的一些属性。
2.1 第一步，构建链表
比如我们有如下一条有序的链表：
我们要查找其中为70的元素，正常情况下，我们需要从头到尾依次遍历匹配，所使用时间复杂度O(n)，同样，插入一个元素并保持链表有序，需要先找到合适的插入位置，再执行插入，时间复杂度也是O(n)。那么如何在此基础上提高查询速度呢，毫无疑问，做索引。
2.2 第二步，添加索引
我们来对原有链表中的一些元素提取出来，作为索引节点。
这样遍历的时候我们先对上层的链表进行遍历，然后获取到元素或者确定元素范围后，再进行下一层的遍历就行了，这样就不用挨个进行比较了，类似于二分查找的方式。
比如查找元素70，上层依次比较10，30，60，80，然后发现70大于60这个节点，小于80这个节点，然后进入下一层进行遍历，这时候只需要遍历60，70即可获取到元素。
所以添加了这层索引节点之后，再查询的时候时间复杂度就是O(n/2)，同理，我们可以不断的增加层数，来降低相应的时间复杂度，直到最上层只有两个元素为止，因为一个节点没有比较的意义。这样最终的时间复杂度就变为了O(logn)。其实这种操作和二分查找类似，通过索引来跳过大量的节点，从而提高查询的效率。
比如再添加一层索引：
最终的结构大致如下：
2.3 插入/删除
正常情况下，上层索引节点与下层链表的元素比例是1:2，但当链表插入或者删除之后，就很难维持这个比例了，所以跳跃表也不强制1:2的比例。
对于添加节点而言，当一些新节点添加之后，哪个节点需要添加为索引节点，以及建几层的索引，都是通过一种抛硬币的形式来决定的。那为什么要使用抛硬币这种方式呢？
因为跳跃表添加和删除的节点都不是固定位置的，很难用一种有效的算法来保证跳跃表的索引始终分配均匀。而随机抛硬币的方式虽然不能保证索引的绝对均匀分布，却可以让大体趋于均匀。
而对于删除节点的话，先遍历找到该节点，然后把该节点和节点对应的所有索引节点一并删除即可。如果某一层索引在删除后只剩下了一个节点，那么这整个一层就可以删掉。
3. 跳跃表总结
上面简单介绍了跳跃表的概念，接下来我们来简单总结下跳跃表：
- 跳跃表是通过多层链表来实现的，每层都是一个有序的链表；
- 跳跃表的最底层包含了所有的元素，其他层保存的是元素的索引节点；
- 跳跃表搜索，删除，插入元素的时间复杂度是O(log N)；
- 跳跃表通过随机的抛硬币方式来决定新插入的节点是否提到上层作为索引节点；
- 跳跃表提高搜索效率的方式是通过空间换时间得到的。
二、ConcurrentSkipListMap
ConcurrentSkipListMap是一个基于skip list实现的线程安全的有序存储的Map，默认情况下根据key的自然顺序进行排序，或者根据在Map进行创建时提供的比较器进行排序。同样，该类不允许key或者value为null。
1. 继承结构
public class ConcurrentSkipListMap<K,V> extends AbstractMap<K,V> implements ConcurrentNavigableMap<K,V>, Cloneable, Serializable { ConcurrentSkipListMap继承了AbstractMap，该抽象类提供了接口的一些常规实现；而继承了ConcurrentNavigableMap接口，该接口可以获取Map的某一部分元素或者子元素；另外，该类支持克隆和序列化操作。
2. 属性
来看一下它的常用属性：
/** * 最底层的头节点的索引 */ private static final Object BASE_HEADER = new Object(); /** * 最顶层头节点索引 */ private transient volatile HeadIndex<K,V> head; /** * 比较器，如果使用自然排序，该值为null */ final Comparator<? super K> comparator; /** key集合 */ private transient KeySet<K> keySet; /** entry集合 */ private transient EntrySet<K,V> entrySet; /** value集合 */ private transient Values<V> values; /** 降序键集合 */ private transient ConcurrentNavigableMap<K,V> descendingMap; 其中比较重要的是比较器，因为ConcurrentSkipListMap会根据key进行排序，如果传为null，则表示是根据元素的自然顺序进行排序。
3. 内部类
ConcurrentSkipListMap中有三个比较重要的内部类，分别是Node，Index，HeadIndex这三个类。Node表示最底层的链表节点，Index类表示基于Node类的索引层，而HeadIndex则是用来维护索引的层次。先来看下Node类：
static final class Node<K,V> { final K key; volatile Object value; volatile Node<K,V> next; } 可以看到，Node这个类中包含了Map的key和value，还包含了一个指向下一节点的指针next，并且这里使用的是单链表结构。然后再来看下Index这个类：
static class Index<K,V> { // Node节点的引用 final Node<K,V> node; // 下层Index的引用 final Index<K,V> down; // 右侧Index的引用 volatile Index<K,V> right; } 可以看出，Index类作为索引节点，共包含了三个属性，一个是Node节点的引用，一个是指向下一层的索引节点的引用，一个是指向右侧索引节点的引用。接下来再来看下HeadIndex这个类：
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; } } 可以看出，HeadIndex继承自Index，扩展了一个level属性，表示当前索引节点Index的层级。
4. 构造方法
ConcurrentSkipListMap共包含了4个构造方法，我们来简单看下：
public ConcurrentSkipListMap() { this.comparator = null; initialize(); } public ConcurrentSkipListMap(Comparator<? super K> comparator) { this.comparator = comparator; initialize(); } 这两个构造方法，一个是默认的方法，表示按照自然顺序进行排序（也就是key必须实现了Comparable接口）；另一个是按照给定的比较器进行排序的构造方法；
public ConcurrentSkipListMap(Map<? extends K, ? extends V> m) { this.comparator = null; initialize(); putAll(m); } 这个构造方法表示从给定的Map构造一个ConcurrentSkipListMap对象，并按照key的自然顺序进行排序；
public ConcurrentSkipListMap(SortedMap<K, ? extends V> m) { this.comparator = m.comparator(); initialize(); buildFromSorted(m); } 这个构造方法表示从给定的Map构造一个ConcurrentSkipListMap对象，而顺序则是按照SortedMap的顺序来进行排序。
这里都调用到了initialize这个方法来初始化对象，我们来看下：
private void initialize() { keySet = null; entrySet = null; values = null; descendingMap = null; // 构造跳跃表的头节点head head = new HeadIndex<K,V>(new Node<K,V>(null, BASE_HEADER, null), null, null, 1); } 可以看到，这里构造了跳跃表的头节点head，构造完成之后，大概如下：
5. 方法
由于该类的方法比较多，所以这里只介绍几个常用的方法。
5.1 put方法
public V put(K key, V value) { // value不能为空 if (value == null) throw new NullPointerException(); return doPut(key, value, false); } 可以看到，put方法内部调用的是doPut方法，在看doPut方法前，我们先来看下该方法中调用到的另一个方法：findPredecessor。
findPredecessor方法表示查询key应该插入位置的前驱节点（如果遇到需要删除的节点，那么进行辅助性删除），从最高层的head节点一直向右方向进行遍历，知道右侧的节点为null或者Node的key大于当前key为止，然后再向下寻找，依次重复该过程，直到down为null，这时候就找到了前驱节点。
private Node<K,V> findPredecessor(Object key, Comparator<? super K> cmp) { if (key == null) throw new NullPointerException(); // don't postpone errors for (;;) { // 从head节点开始遍历，q和r作为临时的head节点和head的右节点 for (Index<K,V> q = head, r = q.right, d;;) { // 如果head节点的右节点存在 if (r != null) { // 获取head节点的右节点 Node<K,V> n = r.node; // 获取右节点的key K k = n.key; // 如果右节点的value为空，说明右侧已经没节点了，该节点已经被删除了 if (n.value == null) { // 通过unlink方法移除该节点 if (!q.unlink(r)) break; // restart // 如果unlink方法返回了false，也就是head的右节点已经有值了 // 那就重新赋值，重新操作 r = q.right; // reread r continue; } // 通过比较器对key进行比较，如果key大于r节点的key，则继续向后遍历 if (cpr(cmp, key, k) > 0) { // 节点后移，q和r整体后移 q = r; r = r.right; continue; } } // 如果q.down == null，表示指针已经到最下层了，直接返回该节点 if ((d = q.down) == null) return q.node; // 进入下层遍历 q = d; r = d.right; } } } 接下来，我们来看下doPut方法：
private V doPut(K key, V value, boolean onlyIfAbsent) { Node<K,V> z; // added node // key不能为空 if (key == null) throw new NullPointerException(); // 获取到比较器 Comparator<? super K> cmp = comparator; // 无限循环 outer: for (;;) { // 先找到应该插入位置的前驱节点，b表示待插入位置的前驱节点，n是前驱节点的后继节点 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) // inconsistent read break; // 如果节点n已经逻辑删除，这里进行辅助性物理删除 if ((v = n.value) == null) { // n is deleted n.helpDelete(b, f); break; } // 如果b已经被删除，结束本层查询 if (b.value == null || v == n) // b is deleted break; // 如果节点key 大于 n节点的key，继续往后遍历 if ((c = cpr(cmp, key, n.key)) > 0) { b = n; n = f; continue; } // 如果节点key与n节点的key相等 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 } // else c < 0; fall through } // 新创建一个节点，next指向n z = new Node<K,V>(key, value, n); // 比较并交换，也就是插入节点；如果竞争失败，重试；成功，则跳出循环 if (!b.casNext(n, z)) break; // restart if lost race to append to b break outer; } } // 获取随机种子 int rnd = ThreadLocalRandom.nextSecondarySeed(); // 测试最低和最高位，用于判断是否需要添加level if ((rnd & 0x80000001) == 0) { // test highest and lowest bits int level = 1, max; // 确定level的级别 while (((rnd = 1) & 1) != 0) ++level; Index<K,V> idx = null; HeadIndex<K,V> h = head; // 如果level小于最大层，就在对应层次以及小于该level的层次进行节点新增处理 if (level <= (max = h.level)) { for (int i = 1; i <= level; ++i) // 为节点生成对应的Index节点，并从下往上依次赋值，并且赋值了Index节点的down节点 idx = new Index<K,V>(z, idx, null); } // 否则，需要新增一层 else { // try to grow by one level level = max + 1; // hold in array and later pick the one to use // 使用数组来保存Index节点 @SuppressWarnings(unchecked)Index<K,V>[] idxs = (Index<K,V>[])new Index<?,?>[level+1]; for (int i = 1; i <= level; ++i) // 从下往上生成Index结点，并赋值down节点，这里数组的第一个值idxs[0]应该是没用到 idxs[i] = idx = new Index<K,V>(z, idx, null); for (;;) { // 保存头节点 h = head; // 保存之前的level 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)) { // 更新head节点，比较并替换 // h赋值为最高层的头节点 h = newh; // idx赋值为之前层级的头结点x，并将level赋值为之前的层级 idx = idxs[level = oldLevel]; break; } } } // find insertion points and splice in // 上述操作只是生成了对应的索引节点，但是并没有将这些节点插入到对应的层之中，下面这些代码是插入Index节点 // 从level层开始操作 splice: for (int insertionLevel = level;;) { // 保存新表的层级 int j = h.level; for (Index<K,V> q = h, r = q.right, t = idx;;) { // 如果头结点或者idx结点为空，跳出这层循环 if (q == null || t == null) break splice; // 如果头节点右侧节点不为空 if (r != null) { Node<K,V> n = r.node; // compare before deletion check avoids needing recheck // key进行比较 int c = cpr(cmp, key, n.key); // 需要删除的节点 if (n.value == null) { if (!q.unlink(r)) break; r = q.right; continue; } // 大于0，向右继续查找 if (c > 0) { q = r; r = r.right; continue; } } // 找到节点进行插入的位置，这里准备进行插入 if (j == insertionLevel) { // 插入，也就是将r结点插入到q与t之间；失败重试 if (!q.link(r, t)) break; // restart 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; } doPut方法内容比较多，我们来梳理下该方法的操作流程：
- 根据key从跳跃表的左上方开始，向右或者向下查找到需要插入位置的前驱Node节点，查找过程中会删除一些已经标记为删除状态的节点；
- 然后判断跳跃表中是否已经存在了该key，如果存在比较并替换；然后生成节点，插入到最底层链表中；
- 然后根据随机值来判断是否生成索引层以及生成索引层的层次；
- 如果需要创建索引层，则判断索引层是否超过最大的level，如果大于，需要创建HeadIndex索引层；否则只需要创建Index索引层即可；
- 从head开始进行遍历，将每一层的新添加的Index插入到对应的位置；
可以看出，该方法还是比较复杂的，大家有时间可以多看两遍。
5.2 get方法
get方法内部调用doGet方法，还有一个getOrDefault方法也是类似的：
public V get(Object key) { return doGet(key); } public V getOrDefault(Object key, V defaultValue) { V v; return (v = doGet(key)) == null ? defaultValue : v; } 接下来主要来看下doGet方法的实现：
private V doGet(Object key) { // 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) // inconsistent read break; // 物理删除对应的节点 if ((v = n.value) == null) { // n is deleted n.helpDelete(b, f); break; } // 如果b已经被删除，结束本层查询 if (b.value == null || v == n) // b is deleted 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; } 相比doPut方法，doGet方法就比较简单了。先找到对应的前驱节点，然后一直向右查找即可，中间如果有某个节点需要删除，顺手删除即可。当然如果发现跳跃表数据结构被其它线程改变，会重新尝试获取其前驱。
5.3 remove方法
remove有两个重载方法，一个是根据key进行删除，另一个是根据key和value一并进行删除，当然内部调用的都是doRemove方法：
public boolean remove(Object key, Object value) { if (key == null) throw new NullPointerException(); return value != null && doRemove(key, value) != null; } public V remove(Object key) { return doRemove(key, null); } 接下来我们来看下doRemove方法：
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; } // 如果b已经被删除，结束本层查询 if (b.value == null || v == n) // b is deleted break; // 如果不存在该key，结束循环 if ((c = cpr(cmp, key, n.key)) < 0) break outer; // 向右进行查找 if (c > 0) { b = n; n = f; continue; } // 判断完key之后，再判断value是否相等，value不相等，退出 if (value != null && !value.equals(v)) break outer; // 将v进行比较并设置为null(逻辑删除)，如果多线程下失败，则重试 if (!n.casValue(v, null)) break; // 先添加一个用于删除标记的节点，然后比较并更新b的next节点 if (!n.appendMarker(f) || !b.casNext(n, f)) findNode(key); // retry via findNode else { // 调用该方法 辅助清除key对应的索引层 findPredecessor(key, cmp); // clean index // 如果该层已经没有节点，删除该层 if (head.right == null) tryReduceLevel(); } @SuppressWarnings(unchecked) V vv = (V)v; return vv; } } return null; } doRemove方法也不是太难，我们来简单梳理下它的流程：
- 先查找到对应的前驱节点，并且查找过程中会删除一些已经标记为删除状态的节点；
- 查询到要删除的节点，通过CAS操作把value设置为null（这样其他线程可以感知到这个节点状态，协助完成删除工作），然后在该节点后面添加一个marker节点作为删除标志位，若添加成功，则将该结点的前驱的后继设置为该结点之前的后继（也就是删除该节点操作），这样可以避免丢失数据；
- 如果该层已经没有了其他节点，调用tryReduceLevel方法移除该层；
这里可能需要说明下，因为ConcurrentSkipListMap是支持并发操作的，因此在删除的时候可能有其他线程在该位置上进行插入，这样有可能导致数据的丢失。在ConcurrentSkipListMap中，会在要删除的节点后面添加一个特殊的节点进行标记，然后再进行整体的删除，如果不进行标记，那么如果正在删除的节点，可能其它线程正在此节点后面添加数据，造成数据丢失。
这个方法中间涉及到了两个小方法，首先是helpDelete方法，用于帮助删除节点的方法，来简单看下：
void helpDelete(Node<K,V> b, Node<K,V> f) { /* * Rechecking links and then doing only one of the * help-out stages per call tends to minimize CAS * interference among helping threads. */ if (f == next && this == b.next) { // 如果没有添加删除标记节点，那么添加删除标记 if (f == null || f.value != f) // not already marked casNext(f, new Node<K,V>(f)); else // 执行删除操作 b.casNext(this, f.next); } } 另外，是减少层级的方法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) && // try to set h.right != null) // recheck casHead(d, h); // try to backout } 这个方法，是针对最上面的三层进行操作，如果最上面的三层HeadIndex的right节点都为空，则减少level的层数，并设置head为之前head的下一层；然后再判断之前的head的right域是否为null，这里是重新校验下，如果为null，则减少层级成功，否则再次将head设置为h（这里为什么是三层，可以看下这个方法的注释介绍）。
5.4 containsValue方法
由于该类的方法特别多，这里我们分析了常用的增加，修改，删除方法外，再随便找个containsValue方法来看下：
public boolean containsValue(Object value) { if (value == null) throw new NullPointerException(); // 查找最底层链表的第一个元素 for (Node<K,V> n = findFirst(); n != null; n = n.next) { // 获取值并进行比较 V v = n.getValidValue(); if (v != null && value.equals(v)) return true; } return false; } 这里调用了findFirst方法获取Map中第一个元素，然后获取到后就一直往右侧进行遍历比较操作，比较简单，再来简单看下findFirst方法：
final Node<K,V> findFirst() { for (Node<K,V> b, n;;) { // 获取第一个保存实际元素的节点，也就是BASE_HEADER节点之后的第一个节点 if ((n = (b = head.node).next) == null) return null; if (n.value != null) return n; // 帮助删除 n.helpDelete(b, n.next); } } 不过可能需要注意的是，ConcurrentSkipListMap的size方法与大多数集合不同，它的size方法不是常量操作，因为它并没有维护一个全局变量来统计元素的个数，而是每次调用该方法的时候都需要去遍历，因此如果在遍历期间该集合发生了修改（也就是多线程情况下），则可能出现不准确的情况。
三、ConcurrentSkipListSet
看完了ConcurrentSkipListMap，我们顺便来看下ConcurrentSkipListSet，这个类比较简单，我们在这里一并看了。ConcurrentSkipListSet是基于ConcurrentSkipListMap实现的并发的Set实现，和通用Set一样，该类元素无法重复，但元素是有序的，可以按照他们的自然顺序进行排序，也可以按照给定的比较器进行排序。
private final ConcurrentNavigableMap<E,Object> m; /** * Constructs a new, empty set that orders its elements according to * their {@linkplain Comparable natural ordering}. */ public ConcurrentSkipListSet() { m = new ConcurrentSkipListMap<E,Object>(); } 内部维护了一个ConcurrentSkipListMap的实现，所有对该Set的操作都是基于对该Map的操作，Set中的元素是保存在Map中的key上，因为ConcurrentSkipListMap中key是不重复的，而该Map对应得value则全是 Boolean.TRUE。我们通过一个例子来看下该类中常用的方法：
// 构造ConcurrentSkipListSet之后，依次添加数据：1 2 3 4 5 6 7 8 9 10 ConcurrentSkipListSet<Integer> skipListSet = new ConcurrentSkipListSet<>(); // 后续以4来进行举例 System.out.println(skipListSet.lower(4)); lower(E) //返回小于给定值的最大值，示例中返回：3 floor(E) //返回小于或者等于给定值的最大值，示例中返回：4 ceiling(E) //返回大于或等于给定值的最小值，示例中返回：4 higher(E) // 返回大于给定值的最小值，示例中返回：5 pollFirst() // 返回并删除最小值，示例中返回：1 pollLast() // 返回并删除最大值，示例中返回：10 first() // 返回集合中的第一个值，也就是最小的：2 last() // 返回集合中的最后一个值，也就是最大的：9 使用场景
这两个类别在几个方面有所不同。
ConcurrentHashMap不保证其操作的运行时作为其合约的一部分。它还允许调整某些加载因子（粗略地说，同时修改它的线程数）。
另一方面，ConcurrentSkipListMap保证在各种操作上的平均O（log（n））性能。它也不支持为并发而调整。ConcurrentSkipListMap还有一些ConcurrentHashMap不支持的操作：ceilingEntry / Key，floorEntry / Key等。它还保持sorting顺序，如果使用的是ConcurrentHashMap，则sorting顺序必须被计算（值得注意）。
基本上，针对不同的使用情况提供了不同的实现。如果您需要快速单键/值对添加和快速单键查找，请使用HashMap。如果您需要更快的顺序遍历，并且可以承担额外的插入成本，请使用SkipListMap。
四、总结
ConcurrentSkipListMap是一个线程安全的基于跳跃表实现的非阻塞的Map，它要求Map中的key和value都不能为null，并且可以通过key来进行排序。内部的实现则是通过多层有序链表来实现的，它使用空间换时间的方式，使得链表也能实现类似二分查找的功能。
而它的应用场景，Redis中的有序集合SortedSet就是基于散列表和跳跃表来实现的。