在学习ConcurrentSkipListMap之前,我们需要来了解一种随机化的数据结构:跳跃表(skip list)。
在我们通常所使用到的数据结构中,使用最多的无外乎就是数组和链表,而数组和链表各自的优缺点也很明显:
数组的优点在于元素的查找,链表的优点在于添加或移除等操作;而对一个有序的数组而言,添加一个元素,可能需要移动大量的元素;而对于一个有序的链表而言,查找一个元素,又不能通过像在数组查找中使用二分查找,只能通过顺序遍历的方式进行。
而为了使链表查询的时候也有一个不错的性能,所以就出现了跳跃表(skip list)这种数据结构。
我们以一个链表查询的小例子来介绍跳跃表,顺便介绍下跳跃表的一些属性。
比如我们有如下一条有序的链表:
我们要查找其中为70的元素,正常情况下,我们需要从头到尾依次遍历匹配,所使用时间复杂度O(n),同样,插入一个元素并保持链表有序,需要先找到合适的插入位置,再执行插入,时间复杂度也是O(n)。那么如何在此基础上提高查询速度呢,毫无疑问,做索引。
我们来对原有链表中的一些元素提取出来,作为索引节点。
这样遍历的时候我们先对上层的链表进行遍历,然后获取到元素或者确定元素范围后,再进行下一层的遍历就行了,这样就不用挨个进行比较了,类似于二分查找的方式。
比如查找元素70,上层依次比较10,30,60,80,然后发现70大于60这个节点,小于80这个节点,然后进入下一层进行遍历,这时候只需要遍历60,70即可获取到元素。
所以添加了这层索引节点之后,再查询的时候时间复杂度就是O(n/2),同理,我们可以不断的增加层数,来降低相应的时间复杂度,直到最上层只有两个元素为止,因为一个节点没有比较的意义。这样最终的时间复杂度就变为了O(logn)。其实这种操作和二分查找类似,通过索引来跳过大量的节点,从而提高查询的效率。
比如再添加一层索引:
最终的结构大致如下:
正常情况下,上层索引节点与下层链表的元素比例是1:2,但当链表插入或者删除之后,就很难维持这个比例了,所以跳跃表也不强制1:2的比例。
对于添加节点而言,当一些新节点添加之后,哪个节点需要添加为索引节点,以及建几层的索引,都是通过一种抛硬币的形式来决定的。那为什么要使用抛硬币这种方式呢?
因为跳跃表添加和删除的节点都不是固定位置的,很难用一种有效的算法来保证跳跃表的索引始终分配均匀。而随机抛硬币的方式虽然不能保证索引的绝对均匀分布,却可以让大体趋于均匀。
而对于删除节点的话,先遍历找到该节点,然后把该节点和节点对应的所有索引节点一并删除即可。如果某一层索引在删除后只剩下了一个节点,那么这整个一层就可以删掉。
上面简单介绍了跳跃表的概念,接下来我们来简单总结下跳跃表:
ConcurrentSkipListMap是一个基于skip list实现的线程安全的有序存储的Map,默认情况下根据key的自然顺序进行排序,或者根据在Map进行创建时提供的比较器进行排序。同样,该类不允许key或者value为null。
public class ConcurrentSkipListMap<K,V> extends AbstractMap<K,V> implements ConcurrentNavigableMap<K,V>, Cloneable, Serializable {
ConcurrentSkipListMap继承了AbstractMap,该抽象类提供了接口的一些常规实现;而继承了ConcurrentNavigableMap接口,该接口可以获取Map的某一部分元素或者子元素;另外,该类支持克隆和序列化操作。
来看一下它的常用属性:
/** * 最底层的头节点的索引 */ 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,则表示是根据元素的自然顺序进行排序。
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的层级。
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,构造完成之后,大概如下:
由于该类的方法比较多,所以这里只介绍几个常用的方法。
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方法内容比较多,我们来梳理下该方法的操作流程:
可以看出,该方法还是比较复杂的,大家有时间可以多看两遍。
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方法就比较简单了。先找到对应的前驱节点,然后一直向右查找即可,中间如果有某个节点需要删除,顺手删除即可。当然如果发现跳跃表数据结构被其它线程改变,会重新尝试获取其前驱。
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方法也不是太难,我们来简单梳理下它的流程:
这里可能需要说明下,因为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(这里为什么是三层,可以看下这个方法的注释介绍)。
由于该类的方法特别多,这里我们分析了常用的增加,修改,删除方法外,再随便找个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方法不是常量操作,因为它并没有维护一个全局变量来统计元素的个数,而是每次调用该方法的时候都需要去遍历,因此如果在遍历期间该集合发生了修改(也就是多线程情况下),则可能出现不准确的情况。
看完了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就是基于散列表和跳跃表来实现的。