这里包括两种容器

ConcurrentSkipListMap
ConcurrentSkipListSet

其中ConcurrentSkipListSet是通过ConcurrentSkipListMap实现的，它包含一个ConcurrentNavigableMap对象m，而m对象实际上是ConcurrentNavigableMap的实现类ConcurrentSkipListMap的实例。

ConcurrentSkipListMap中的元素是key-value键值对；而ConcurrentSkipListSet是集合，它只用到了ConcurrentSkipListMap中的key，其value是一个空的Object。

public class ConcurrentSkipListSet<E>
    extends AbstractSet<E>
    implements NavigableSet<E>, Cloneable, java.io.Serializable {
    private final ConcurrentNavigableMap<E,Object> m;

    public ConcurrentSkipListSet() {
        m = new ConcurrentSkipListMap<E,Object>();
    }
}

因此这里只分析ConcurrentSkipListMap的源代码。

1.ConcurrentSkipListMap

ConcurrentSkipListMap和TreeMap类似，它们虽然都是有序的哈希表。但是，第一，它们的线程安全机制不同，TreeMap是非线程安全的，而ConcurrentSkipListMap是线程安全的。第二，ConcurrentSkipListMap是通过跳表实现的，而TreeMap是通过红黑树实现的。

ConcurrentSkipListMap的数据结构，如下图所示：

跳表分为许多层(level)，每一层都可以看作是数据的索引，这些索引的意义就是加快跳表查找数据速度。每一层的数据都是有序的，上一层数据是下一层数据的子集，并且第一层(level 1)包含了全部的数据；层次越高，跳跃性越大，包含的数据越少。跳表包含一个表头，它查找数据时，是从上往下，从左往右进行查找。

先以数据“7,14,21,32,37,71,85”序列为例，来对跳表进行简单说明。在跳表中查找“32”节点路径如下图所示：

public class ConcurrentSkipListMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentNavigableMap<K,V>,
               Cloneable,
               java.io.Serializable {
    //head是跳表的表头
    private transient volatile HeadIndex<K,V> head;
}

static class Index<K,V> {
    final Node<K,V> node;       //哈希表节点node
    final Index<K,V> down;      //下索引的指针
    volatile Index<K,V> right;  //右索引的指针
}

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;
    }
}

static final class Node<K,V> {
    final K key;
    volatile Object value;
    volatile Node<K,V> next;
}

下面从ConcurrentSkipListMap的添加(put)，删除(remove)，获取(get)这3个方面对它进行分析：

1.1 添加

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

private V doPut(K kkey, V value, boolean onlyIfAbsent) {
    Comparable<? super K> key = comparable(kkey);
    for (;;) {
        Node<K,V> b = findPredecessor(key); //找到key的前继节点
        Node<K,V> n = b.next;   //key的前继节点的后继节点，即 插入节点 的“后继节点”
        for (;;) {
            if (n != null) {
                Node<K,V> f = n.next;
                //如果两次获得的b.next不是相同的Node，就跳转到外层循环，重新遍历。
                if (n != b.next)               // inconsistent read
                    break;;
                Object v = n.value;
                //当n的值为null(意味着其它线程删除了n)；此时删除b的下一个节点，然后跳转到”外层for循环“，重新遍历。
                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;
                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);
            if (!b.casNext(n, z))   // 设置“b的后继节点”为z
                break;         // restart if lost race to append to b，多线程情况下，break才可能发生(其它线程对b进行了操作)

            // 随机获取一个level，然后在“第1层”到“第level层”的链表中都插入新建节点
            int level = randomLevel();
            if (level > 0)
                insertIndex(z, level);
            return null;
        }
    }
}

doPut()的作用就是将键值对添加到“跳表”中,其主干部分(即单纯的只考虑“单线程的情况下，将key-value添加到跳表中”，即忽略“多线程相关的内容”),它的流程如下：

第1步：找到“插入位置”。找到key的前去节点(b)和key的后继节点(n)；key是要插入节点的键;
第2步：新建并插入节点。新建节点z(key对应的节点)，并将新节点z插入到“跳表”中(设置“b的后继节点为z”，“z的后继节点为n”);
第3步：更新跳表。即随机获取一个level，然后在“跳表”的第1层～第level层之间，每一层都插入节点z；在第level层之上就不再插入节点了。若level数值大于“跳表的层次”，则新建一层;

主干代码大致如下：

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的后继节点
        Node<K,V> n = b.next;
        for (;;) {
            // 新建节点(对应是“要被插入的键值对”)
            Node<K,V> z = new Node<K,V>(kkey, value, n);
            // 设置“b的后继节点”为z
            b.casNext(n, z);

            // 随机获取一个level，然后在“第1层”到“第level层”的链表中都插入新建节点
            int level = randomLevel();
            if (level > 0)
                insertIndex(z, level);
            return null;
        }
    }
}

1.2 删除

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

final V doRemove(Object okey, Object value) {
        Comparable<? super K> key = comparable(okey);
        for (;;) {
            Node<K,V> b = findPredecessor(key); //前置节点
            Node<K,V> n = b.next;
            for (;;) {
                if (n == null)
                    return null;
                Node<K,V> f = n.next;
                if (n != b.next)                    // inconsistent read
                    break;
                Object v = n.value;
                if (v == null) {                    // n is deleted
                    n.helpDelete(b, f);
                    break;
                }
                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;
                }
                if (value != null && !value.equals(v))
                    return null;
                if (!n.casValue(v, null))// 设置b的后继节点为null
                    break;
                if (!n.appendMarker(f) || !b.casNext(n, f))
                    findNode(key);                  // Retry via findNode
                else {
                    findPredecessor(key);           // Clean index
                    if (head.right == null)
                        tryReduceLevel();
                }
                return (V)v;
            }
        }
    }

下面是删除跳表中键值对的主干步骤：

第1步：找到被删除节点的位置;找到key的前继节点(b)，key所对应的节点(n)，n的后继节点f；key是要删除节点的键;
第2步：删除节点;将key所对应的节点n从跳表中移除，将b的后继节点设为f;
第3步：更新跳表;遍历跳表，删除每一层的key节点(如果存在的话)。删除导致某一层为空之后，需要删除这一层;

主干部分的doRemove()的代码大致如下：

final V doRemove(Object okey, Object value) {
    Comparable<? super K> key = comparable(okey);
    for (;;) {
        // 找到key的前继节点b
        Node<K,V> b = findPredecessor(key);
        // b-->n,即n为b的后继节点，也即n节点为key的节点
        Node<K,V> n = b.next;
        for (;;) {
            // f是n的后继节点
            Node<K,V> f = n.next;
            // 设置n节点为null
            n.casValue(v, null);
            // 设置b的后继节点为f
            b.casNext(n, f);
            // 清除跳表中每一层的key节点
            findPredecessor(key);
            // 如果“表头的右索引为空”，则将“跳表的层次”-1
            if (head.right == null)
                tryReduceLevel();
            return (V)v;
        }
    }
}

1.3 获取

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

private V doGet(Object okey) {
    Comparable<? super K> key = comparable(okey);
    Node<K,V> bound = null;
    Index<K,V> q = head;    //待比较节点
    Index<K,V> r = q.right; //待比较的右边节点
    Node<K,V> n;
    K k;
    int c;
    for (;;) {
        Index<K,V> d;
        // 往右遍历
        if (r != null && (n = r.node) != bound && (k = n.key) != null) {
            if ((c = key.compareTo(k)) > 0) {
                q = r;
                r = r.right;
                continue;
            } else if (c == 0) {
                Object v = n.value;
                return (v != null)? (V)v : getUsingFindNode(key);
            } else
                bound = n;
        }

        // 往下遍历
        if ((d = q.down) != null) {
            q = d;
            r = d.right;
        } else
            break;
    }

    // 同一层逐个往后便利
    for (n = q.node.next;  n != null; n = n.next) {
        if ((k = n.key) != null) {
            if ((c = key.compareTo(k)) == 0) {
                Object v = n.value;
                return (v != null)? (V)v : getUsingFindNode(key);
            } else if (c < 0)
                break;
        }
    }
    return null;
}

2.ShipList的简单实现

很早之前写过一个简单的SkipList，贴在百度空间： http://hi.baidu.com/nicker2010/item/be641424bb0963c8ef10f1dc

xiaobaoqiu Blog

Think More, Code Less

Java并发容器之SkipList

1.ConcurrentSkipListMap

1.1 添加

1.2 删除

1.3 获取

2.ShipList的简单实现