ConcurrentSkipListMap源码分析

概述

ConcurrentSkipListMap是一个线程安全的有序的哈希表，并发安全主要由CAS来实现。内部的使用了跳表这种数据结构。ConcurrentSkipListSet的底层是基于ConcurrentSkipListMap实现的。

跳表（Skip List）

跳表可以做到比较稳定的插入，查询与删除。理论插入查询删除的算法时间复杂度为O(logN).
跳表其实是从链表之上改进而来的。我们知道在链表中查询，插入的时间复杂度都为O(n).如果我们可以在多个节点之间跳跃，就可以提高效率。

链表的结构如下；

改进后得到：

如果层次更多的话，那么结构如下；

跳表的查询：
假如我们要查询11.从最上面一层出发，发现11大于5小于13，那么确定了大致区间。进入第二层，发现11大于9小于13，区间缩小。进入第三层，依次查找，最终找到11这个节点。

插入与查询的过程也非常的类似，首先找到在最底层合适的位置，然后再随机是否向上拓展。

删除同样也需要查找，然后再从下至上依次删除。

ConcurrentSkipListMap源码分析

重要属性

/**
 * 跳表最底一层的链表的头节点。
 */
private static final Object BASE_HEADER = new Object();

/**
 * 跳表最高层的头节点
 */
private transient volatile HeadIndex<K,V> head;

ConcurrentSkipListMap的跳表实现：

数据存储由三个内部类实现：
Node:存储键值对，单向链表节点。
Index:跳表中的索引节点，包含了向右的指针和向下的指针，和节点。
HeadIndex：跳表的头，继承至Index，包含了层次信息。

重要方法

put方法

public V put(K key, V value) {
    if (value == null)//不支持null键
        throw new NullPointerException();
    return doPut(key, value, false);
}

从put方法中，我们可以知道ConcurrentSkipListMap不支持null键。

实际的put工作由doPut方法完成。

private V doPut(K key, V value, boolean onlyIfAbsent) {
       Node<K,V> z;             // added node
       if (key == null) //不支持空键
           throw new NullPointerException();
       Comparator<? super K> cmp = comparator;
       outer: for (;;) {
       //找到指定key节点的前驱节点
           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;
                   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) {
                   //当前key大于n.key，继续向后查找
                       b = n;
                       n = f;
                       continue;
                   }
                   if (c == 0) {
                       //更新value，返回更新前的值
                       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
               }

               //新建一个节点，插入的哦b和b.next之间
               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();
       //使用随机数来决定是否更新层级
       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;
           //构建index的逻辑
           if (level <= (max = h.level)) { //不需要增加层级
               for (int i = 1; i <= level; ++i)
                   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
               //构建一个长度为level+1的index数组
               @SuppressWarnings("unchecked")Index<K,V>[] idxs =
                   (Index<K,V>[])new Index<?,?>[level+1];
               //从下至上构建HeadIndex
               for (int i = 1; i <= level; ++i)
                   idxs[i] = idx = new Index<K,V>(z, idx, null);
               for (;;) { //自旋
                   h = head;
                   //保存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;
                   }
               }
           }
           // find insertion points and splice in 插入index
           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;
                       // compare before deletion check avoids needing recheck
                       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; // 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;
   }

这个方法的实现非常的复杂。但大致可以分为两个步骤；

通过自旋查找索引位置，更新或插入给定的节点元素。在遍历查找的过程中，也会帮助清楚已经删除的节点。具体的流程如下：

首先通过findPredecessor方法，从最底一层找到key节点的前驱节点b，从这个节点开始先后查找合适位置插入。
如果在查找的过程中，发现了已删除的节点，那么会调用helpDelete方法帮助清除节点。

通过随机的方式，确定是否更新跳表层级。随机的过程大致如下：首先生成一个随机数rnd,如果rnd为正偶数，那么就会进行下一步的判断，计算rnd从第2位开始有多少个连续的1，如果连续1的数量小于等于旧表层级，则不需要增加跳表层级，只需要更新index,否则旧需要更新跳表层级。

remove方法

1
2
3

public V remove(Object key) {
    return doRemove(key, null);
}

remove实际上是调用了doRemove方法。

final V doRemove(Object key, Object value) {
       if (key == null)
           throw new NullPointerException();
       Comparator<? super K> cmp = comparator;
       outer: for (;;) {
       //找到指定key的前驱节点
           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)) //找打了指定的节点将value置null
                   break;
                   //添加删除标识，彻底从链表上删除
               if (!n.appendMarker(f) || !b.casNext(n, f))
                   findNode(key);                  // retry via findNode
               else {
               //删除n节点对应的index
                   findPredecessor(key, cmp);      // clean index
                   if (head.right == null)
                   //减少跳表的层级
                       tryReduceLevel();
               }
               @SuppressWarnings("unchecked") V vv = (V)v;
               return vv; //返回对应的value
           }
       }
       return null;
   }

整个方法的流程大致如下：

首先找到需要删除节点的前系欸但，如果在查找的过程中发现已经删除的节点，那么旧帮助清除节点
子啊找打需要删除的节点时，不会理解移除它，而是会通过CAS添加一个删除标识，然后再利用CAS来解除链接，如果途中CAS执行失败，那么就会调用findNode来删除有删除标记的节点。
最后检查head.right如果已经被移除了，那么就会调用tryReduceLevel方法尝试对跳表进行降级操作（只有层级大于三才可以降级）。

get方法

1
2
3

public V get(Object key) {
    return doGet(key);
}

我们可以发现，实际完成get操作的是doGet方法。

private V doGet(Object key) {
        if (key == null)
            throw new NullPointerException();
        Comparator<? super K> cmp = comparator;
        outer: for (;;) {
        //从最底层查找指定key节点的前驱节点
            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已经被删除了，帮助清除已经删除的节点
                    n.helpDelete(b, f);
                    break;
                }
                if (b.value == null || v == n)  // b is deleted
                    break;
                if ((c = cpr(cmp, key, n.key)) == 0) {//检查k是否相等
                    @SuppressWarnings("unchecked") V vv = (V)v;
                    return vv;
                }
                if (c < 0)
                    break outer;
                //未找到合适节点，继续向后查找
                b = n;
                n = f;
            }
        }
        return null;
    }