死磕 java集合之ConcurrentSkipListMap源碼分析——發(fā)現(xiàn)個bug

前情提要

點擊鏈接查看“跳表”詳細(xì)介紹。

拜托，面試別再問我跳表了！

簡介

跳表是一個隨機(jī)化的數(shù)據(jù)結(jié)構(gòu)，實質(zhì)就是一種可以進(jìn)行二分查找的有序鏈表。

跳表在原有的有序鏈表上面增加了多級索引，通過索引來實現(xiàn)快速查找。

跳表不僅能提高搜索性能，同時也可以提高插入和刪除操作的性能。

存儲結(jié)構(gòu)

跳表在原有的有序鏈表上面增加了多級索引，通過索引來實現(xiàn)快速查找。

源碼分析

主要內(nèi)部類

內(nèi)部類跟存儲結(jié)構(gòu)結(jié)合著來看，大概能預(yù)測到代碼的組織方式。

// 數(shù)據(jù)節(jié)點，典型的單鏈表結(jié)構(gòu)
static final class Node<K,V> {
    final K key;
    // 注意：這里value的類型是Object，而不是V
    // 在刪除元素的時候value會指向當(dāng)前元素本身
    volatile Object value;
    volatile Node<K,V> next;

    Node(K key, Object value, Node<K,V> next) {
        this.key = key;
        this.value = value;
        this.next = next;
    }

    Node(Node<K,V> next) {
        this.key = null;
        this.value = this; // 當(dāng)前元素本身(marker)
        this.next = next;
    }
}

// 索引節(jié)點，存儲著對應(yīng)的node值，及向下和向右的索引指針
static class Index<K,V> {
    final Node<K,V> node;
    final Index<K,V> down;
    volatile Index<K,V> right;

    Index(Node<K,V> node, Index<K,V> down, Index<K,V> right) {
        this.node = node;
        this.down = down;
        this.right = right;
    }
}

// 頭索引節(jié)點，繼承自Index，并擴(kuò)展一個level字段，用于記錄索引的層級
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）Node，數(shù)據(jù)節(jié)點，存儲數(shù)據(jù)的節(jié)點，典型的單鏈表結(jié)構(gòu)；

（2）Index，索引節(jié)點，存儲著對應(yīng)的node值，及向下和向右的索引指針；

（3）HeadIndex，頭索引節(jié)點，繼承自Index，并擴(kuò)展一個level字段，用于記錄索引的層級；

構(gòu)造方法

public ConcurrentSkipListMap() {
    this.comparator = null;
    initialize();
}

public ConcurrentSkipListMap(Comparator<? super K> comparator) {
    this.comparator = comparator;
    initialize();
}

public ConcurrentSkipListMap(Map<? extends K, ? extends V> m) {
    this.comparator = null;
    initialize();
    putAll(m);
}

public ConcurrentSkipListMap(SortedMap<K, ? extends V> m) {
    this.comparator = m.comparator();
    initialize();
    buildFromSorted(m);
}

四個構(gòu)造方法里面都調(diào)用了initialize()這個方法，那么，這個方法里面有什么呢？

private static final Object BASE_HEADER = new Object();

private void initialize() {
    keySet = null;
    entrySet = null;
    values = null;
    descendingMap = null;
    // Node(K key, Object value, Node<K,V> next)
    // HeadIndex(Node<K,V> node, Index<K,V> down, Index<K,V> right, int level)
    head = new HeadIndex<K,V>(new Node<K,V>(null, BASE_HEADER, null),
                              null, null, 1);
}

可以看到，這里初始化了一些屬性，并創(chuàng)建了一個頭索引節(jié)點，里面存儲著一個數(shù)據(jù)節(jié)點，這個數(shù)據(jù)節(jié)點的值是空對象，且它的層級是1。

所以，初始化的時候，跳表中只有一個頭索引節(jié)點，層級是1，數(shù)據(jù)節(jié)點是一個空對象，down和right都是null。

通過內(nèi)部類的結(jié)構(gòu)我們知道，一個頭索引指針包含node, down, right三個指針，為了便于理解，我們把指向node的指針用虛線表示，其它兩個用實線表示，也就是虛線不是表明方向的。

添加元素

通過【拜托，面試別再問我跳表了！】中的分析，我們知道跳表插入元素的時候會通過拋硬幣的方式?jīng)Q定出它需要的層級，然后找到各層鏈中它所在的位置，最后通過單鏈表插入的方式把節(jié)點及索引插入進(jìn)去來實現(xiàn)的。

那么，ConcurrentSkipList中是這么做的嗎？讓我們一起來探個究竟：

public V put(K key, V value) {
    // 不能存儲value為null的元素
    // 因為value為null標(biāo)記該元素被刪除（后面會看到）
    if (value == null)
        throw new NullPointerException();

    // 調(diào)用doPut()方法添加元素
    return doPut(key, value, false);
}

private V doPut(K key, V value, boolean onlyIfAbsent) {
    // 添加元素后存儲在z中
    Node<K,V> z;             // added node
    // key也不能為null
    if (key == null)
        throw new NullPointerException();
    Comparator<? super K> cmp = comparator;

    // Part I：找到目標(biāo)節(jié)點的位置并插入
    // 這里的目標(biāo)節(jié)點是數(shù)據(jù)節(jié)點，也就是最底層的那條鏈
    // 自旋
    outer: for (;;) {
        // 尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點，存儲在b中，b=before
        // 并把b的下一個數(shù)據(jù)節(jié)點存儲在n中，n=next
        // 為了便于描述，我這里把b叫做當(dāng)前節(jié)點，n叫做下一個節(jié)點
        for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
            // 如果下一個節(jié)點不為空
            // 就拿其key與目標(biāo)節(jié)點的key比較，找到目標(biāo)節(jié)點應(yīng)該插入的位置
            if (n != null) {
                // v=value，存儲節(jié)點value值
                // c=compare，存儲兩個節(jié)點比較的大小
                Object v; int c;
                // n的下一個數(shù)據(jù)節(jié)點，也就是b的下一個節(jié)點的下一個節(jié)點（孫子節(jié)點）
                Node<K,V> f = n.next;
                // 如果n不為b的下一個節(jié)點
                // 說明有其它線程修改了數(shù)據(jù)，則跳出內(nèi)層循環(huán)
                // 也就是回到了外層循環(huán)自旋的位置，從頭來過
                if (n != b.next)               // inconsistent read
                    break;
                // 如果n的value值為空，說明該節(jié)點已刪除，協(xié)助刪除節(jié)點
                if ((v = n.value) == null) {   // n is deleted
                    // todo 這里為啥會協(xié)助刪除？后面講
                    n.helpDelete(b, f);
                    break;
                }
                // 如果b的值為空或者v等于n，說明b已被刪除
                // 這時候n就是marker節(jié)點，那b就是被刪除的那個
                if (b.value == null || v == n) // b is deleted
                    break;
                // 如果目標(biāo)key與下一個節(jié)點的key大
                // 說明目標(biāo)元素所在的位置還在下一個節(jié)點的后面
                if ((c = cpr(cmp, key, n.key)) > 0) {
                    // 就把當(dāng)前節(jié)點往后移一位
                    // 同樣的下一個節(jié)點也往后移一位
                    // 再重新檢查新n是否為空，它與目標(biāo)key的關(guān)系
                    b = n;
                    n = f;
                    continue;
                }
                // 如果比較時發(fā)現(xiàn)下一個節(jié)點的key與目標(biāo)key相同
                // 說明鏈表中本身就存在目標(biāo)節(jié)點
                if (c == 0) {
                    // 則用新值替換舊值，并返回舊值（onlyIfAbsent=false）
                    if (onlyIfAbsent || n.casValue(v, value)) {
                        @SuppressWarnings("unchecked") V vv = (V)v;
                        return vv;
                    }
                    // 如果替換舊值時失敗，說明其它線程先一步修改了值，從頭來過
                    break; // restart if lost race to replace value
                }
                // 如果c<0，就往下走，也就是找到了目標(biāo)節(jié)點的位置
                // else c < 0; fall through
            }

            // 有兩種情況會到這里
            // 一是到鏈表尾部了，也就是n為null了
            // 二是找到了目標(biāo)節(jié)點的位置，也就是上面的c<0

            // 新建目標(biāo)節(jié)點，并賦值給z
            // 這里把n作為新節(jié)點的next
            // 如果到鏈表尾部了，n為null，這毫無疑問
            // 如果c<0，則n的key比目標(biāo)key大，相妝于在b和n之間插入目標(biāo)節(jié)點z
            z = new Node<K,V>(key, value, n);
            // 原子更新b的下一個節(jié)點為目標(biāo)節(jié)點z
            if (!b.casNext(n, z))
                // 如果更新失敗，說明其它線程先一步修改了值，從頭來過
                break;         // restart if lost race to append to b
            // 如果更新成功，跳出自旋狀態(tài)
            break outer;
        }
    }

    // 經(jīng)過Part I，目標(biāo)節(jié)點已經(jīng)插入到有序鏈表中了

    // Part II：隨機(jī)決定是否需要建立索引及其層次，如果需要則建立自上而下的索引

    // 取個隨機(jī)數(shù)
    int rnd = ThreadLocalRandom.nextSecondarySeed();
    // 0x80000001展開為二進(jìn)制為10000000000000000000000000000001
    // 只有兩頭是1
    // 這里(rnd & 0x80000001) == 0
    // 相當(dāng)于排除了負(fù)數(shù)（負(fù)數(shù)最高位是1），排除了奇數(shù)（奇數(shù)最低位是1）
    // 只有最高位最低位都不為1的數(shù)跟0x80000001做&操作才會為0
    // 也就是正偶數(shù)
    if ((rnd & 0x80000001) == 0) { // test highest and lowest bits
        // 默認(rèn)level為1，也就是只要到這里了就會至少建立一層索引
        int level = 1, max;
        // 隨機(jī)數(shù)從最低位的第二位開始，有幾個連續(xù)的1則level就加幾
        // 因為最低位肯定是0，正偶數(shù)嘛
        // 比如，1100110，level就加2
        while (((rnd >>>= 1) & 1) != 0)
            ++level;

        // 用于記錄目標(biāo)節(jié)點建立的最高的那層索引節(jié)點
        Index<K,V> idx = null;
        // 取頭索引節(jié)點（這是最高層的頭索引節(jié)點）
        HeadIndex<K,V> h = head;
        // 如果生成的層數(shù)小于等于當(dāng)前最高層的層級
        // 也就是跳表的高度不會超過現(xiàn)有高度
        if (level <= (max = h.level)) {
            // 從第一層開始建立一條豎直的索引鏈表
            // 這條鏈表使用down指針連接起來
            // 每個索引節(jié)點里面都存儲著目標(biāo)節(jié)點這個數(shù)據(jù)節(jié)點
            // 最后idx存儲的是這條索引鏈表的最高層節(jié)點
            for (int i = 1; i <= level; ++i)
                idx = new Index<K,V>(z, idx, null);
        }
        else { // try to grow by one level
            // 如果新的層數(shù)超過了現(xiàn)有跳表的高度
            // 則最多只增加一層
            // 比如現(xiàn)在只有一層索引，那下一次最多增加到兩層索引，增加多了也沒有意義
            level = max + 1; // hold in array and later pick the one to use
            // idxs用于存儲目標(biāo)節(jié)點建立的豎起索引的所有索引節(jié)點
            // 其實這里直接使用idx這個最高節(jié)點也是可以完成的
            // 只是用一個數(shù)組存儲所有節(jié)點要方便一些
            // 注意，這里數(shù)組0號位是沒有使用的
            @SuppressWarnings("unchecked")Index<K,V>[] idxs =
                    (Index<K,V>[])new Index<?,?>[level+1];
            // 從第一層開始建立一條豎的索引鏈表（跟上面一樣，只是這里順便把索引節(jié)點放到數(shù)組里面了）
            for (int i = 1; i <= level; ++i)
                idxs[i] = idx = new Index<K,V>(z, idx, null);

            // 自旋
            for (;;) {
                // 舊的最高層頭索引節(jié)點
                h = head;
                // 舊的最高層級
                int oldLevel = h.level;
                // 再次檢查，如果舊的最高層級已經(jīng)不比新層級矮了
                // 說明有其它線程先一步修改了值，從頭來過
                if (level <= oldLevel) // lost race to add level
                    break;
                // 新的最高層頭索引節(jié)點
                HeadIndex<K,V> newh = h;
                // 頭節(jié)點指向的數(shù)據(jù)節(jié)點
                Node<K,V> oldbase = h.node;
                // 超出的部分建立新的頭索引節(jié)點
                for (int j = oldLevel+1; j <= level; ++j)
                    newh = new HeadIndex<K,V>(oldbase, newh, idxs[j], j);
                // 原子更新頭索引節(jié)點
                if (casHead(h, newh)) {
                    // h指向新的最高層頭索引節(jié)點
                    h = newh;
                    // 把level賦值為舊的最高層級的
                    // idx指向的不是最高的索引節(jié)點了
                    // 而是與舊最高層平齊的索引節(jié)點
                    idx = idxs[level = oldLevel];
                    break;
                }
            }
        }

        // 經(jīng)過上面的步驟，有兩種情況
        // 一是沒有超出高度，新建一條目標(biāo)節(jié)點的索引節(jié)點鏈
        // 二是超出了高度，新建一條目標(biāo)節(jié)點的索引節(jié)點鏈，同時最高層頭索引節(jié)點同樣往上長

        // Part III：將新建的索引節(jié)點（包含頭索引節(jié)點）與其它索引節(jié)點通過右指針連接在一起

        // 這時level是等于舊的最高層級的，自旋
        splice: for (int insertionLevel = level;;) {
            // h為最高頭索引節(jié)點
            int j = h.level;

            // 從頭索引節(jié)點開始遍歷
            // 為了方便，這里叫q為當(dāng)前節(jié)點，r為右節(jié)點，d為下節(jié)點，t為目標(biāo)節(jié)點相應(yīng)層級的索引
            for (Index<K,V> q = h, r = q.right, t = idx;;) {
                // 如果遍歷到了最右邊，或者最下邊，
                // 也就是遍歷到頭了，則退出外層循環(huán)
                if (q == null || t == null)
                    break splice;
                // 如果右節(jié)點不為空
                if (r != null) {
                    // n是右節(jié)點的數(shù)據(jù)節(jié)點，為了方便，這里直接叫右節(jié)點的值
                    Node<K,V> n = r.node;
                    // 比較目標(biāo)key與右節(jié)點的值
                    int c = cpr(cmp, key, n.key);
                    // 如果右節(jié)點的值為空了，則表示此節(jié)點已刪除
                    if (n.value == null) {
                        // 則把右節(jié)點刪除
                        if (!q.unlink(r))
                            // 如果刪除失敗，說明有其它線程先一步修改了，從頭來過
                            break;
                        // 刪除成功后重新取右節(jié)點
                        r = q.right;
                        continue;
                    }
                    // 如果比較c>0，表示目標(biāo)節(jié)點還要往右
                    if (c > 0) {
                        // 則把當(dāng)前節(jié)點和右節(jié)點分別右移
                        q = r;
                        r = r.right;
                        continue;
                    }
                }

                // 到這里說明已經(jīng)到當(dāng)前層級的最右邊了
                // 這里實際是會先走第二個if

                // 第一個if
                // j與insertionLevel相等了
                // 實際是先走的第二個if，j自減后應(yīng)該與insertionLevel相等
                if (j == insertionLevel) {
                    // 這里是真正連右指針的地方
                    if (!q.link(r, t))
                        // 連接失敗，從頭來過
                        break; // restart
                    // t節(jié)點的值為空，可能是其它線程刪除了這個元素
                    if (t.node.value == null) {
                        // 這里會去協(xié)助刪除元素
                        findNode(key);
                        break splice;
                    }
                    // 當(dāng)前層級右指針連接完畢，向下移一層繼續(xù)連接
                    // 如果移到了最下面一層，則說明都連接完成了，退出外層循環(huán)
                    if (--insertionLevel == 0)
                        break splice;
                }

                // 第二個if
                // j先自減1，再與兩個level比較
                // j、insertionLevel和t(idx)三者是對應(yīng)的，都是還未把右指針連好的那個層級
                if (--j >= insertionLevel && j < level)
                    // t往下移
                    t = t.down;

                // 當(dāng)前層級到最右邊了
                // 那只能往下一層級去走了
                // 當(dāng)前節(jié)點下移
                // 再取相應(yīng)的右節(jié)點
                q = q.down;
                r = q.right;
            }
        }
    }
    return null;
}

// 尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點
private Node<K,V> findPredecessor(Object key, Comparator<? super K> cmp) {
    // key不能為空
    if (key == null)
        throw new NullPointerException(); // don't postpone errors
    // 自旋
    for (;;) {
        // 從最高層頭索引節(jié)點開始查找，先向右，再向下
        // 直到找到目標(biāo)位置之前的那個索引
        for (Index<K,V> q = head, r = q.right, d;;) {
            // 如果右節(jié)點不為空
            if (r != null) {
                // 右節(jié)點對應(yīng)的數(shù)據(jù)節(jié)點，為了方便，我們叫右節(jié)點的值
                Node<K,V> n = r.node;
                K k = n.key;
                // 如果右節(jié)點的value為空
                // 說明其它線程把這個節(jié)點標(biāo)記為刪除了
                // 則協(xié)助刪除
                if (n.value == null) {
                    if (!q.unlink(r))
                        // 如果刪除失敗
                        // 說明其它線程先刪除了，從頭來過
                        break;           // restart
                    // 刪除之后重新讀取右節(jié)點
                    r = q.right;         // reread r
                    continue;
                }
                // 如果目標(biāo)key比右節(jié)點還大，繼續(xù)向右尋找
                if (cpr(cmp, key, k) > 0) {
                    // 往右移
                    q = r;
                    // 重新取右節(jié)點
                    r = r.right;
                    continue;
                }
                // 如果c<0，說明不能再往右了
            }
            // 到這里說明當(dāng)前層級已經(jīng)到最右了
            // 兩種情況：一是r==null，二是c<0
            // 再從下一級開始找

            // 如果沒有下一級了，就返回這個索引對應(yīng)的數(shù)據(jù)節(jié)點
            if ((d = q.down) == null)
                return q.node;

            // 往下移
            q = d;
            // 重新取右節(jié)點
            r = d.right;
        }
    }
}

// Node.class中的方法，協(xié)助刪除元素
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.
     */
    // 這里的調(diào)用者this==n，三者關(guān)系是b->n->f
    if (f == next && this == b.next) {
        // 將n的值設(shè)置為null后，會先把n的下個節(jié)點設(shè)置為marker節(jié)點
        // 這個marker節(jié)點的值是它自己
        // 這里如果不是它自己說明marker失敗了，重新marker
        if (f == null || f.value != f) // not already marked
            casNext(f, new Node<K,V>(f));
        else
            // marker過了，就把b的下個節(jié)點指向marker的下個節(jié)點
            b.casNext(this, f.next);
    }
}

// Index.class中的方法，刪除succ節(jié)點
final boolean unlink(Index<K,V> succ) {
    // 原子更新當(dāng)前節(jié)點指向下一個節(jié)點的下一個節(jié)點
    // 也就是刪除下一個節(jié)點
    return node.value != null && casRight(succ, succ.right);
}

// Index.class中的方法，在當(dāng)前節(jié)點與succ之間插入newSucc節(jié)點
final boolean link(Index<K,V> succ, Index<K,V> newSucc) {
    // 在當(dāng)前節(jié)點與下一個節(jié)點中間插入一個節(jié)點
    Node<K,V> n = node;
    // 新節(jié)點指向當(dāng)前節(jié)點的下一個節(jié)點
    newSucc.right = succ;
    // 原子更新當(dāng)前節(jié)點的下一個節(jié)點指向新節(jié)點
    return n.value != null && casRight(succ, newSucc);
}

我們這里把整個插入過程分成三個部分：

Part I：找到目標(biāo)節(jié)點的位置并插入

（1）這里的目標(biāo)節(jié)點是數(shù)據(jù)節(jié)點，也就是最底層的那條鏈；

（2）尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點（數(shù)據(jù)節(jié)點都是在最底層的鏈表上）；

（3）從這個數(shù)據(jù)節(jié)點開始往后遍歷，直到找到目標(biāo)節(jié)點應(yīng)該插入的位置；

（4）如果這個位置有元素，就更新其值（onlyIfAbsent=false）；

（5）如果這個位置沒有元素，就把目標(biāo)節(jié)點插入；

（6）至此，目標(biāo)節(jié)點已經(jīng)插入到最底層的數(shù)據(jù)節(jié)點鏈表中了；

Part II：隨機(jī)決定是否需要建立索引及其層次，如果需要則建立自上而下的索引

（1）取個隨機(jī)數(shù)rnd，計算(rnd & 0x80000001)；

（2）如果不等于0，結(jié)束插入過程，也就是不需要創(chuàng)建索引，返回；

（3）如果等于0，才進(jìn)入創(chuàng)建索引的過程（只要正偶數(shù)才會等于0）；

（4）計算while (((rnd >>>= 1) & 1) != 0)，決定層級數(shù)，level從1開始；

（5）如果算出來的層級不高于現(xiàn)有最高層級，則直接建立一條豎直的索引鏈表（只有down有值），并結(jié)束Part II；

（6）如果算出來的層級高于現(xiàn)有最高層級，則新的層級只能比現(xiàn)有最高層級多1；

（7）同樣建立一條豎直的索引鏈表（只有down有值）；

（8）將頭索引也向上增加到相應(yīng)的高度，結(jié)束Part II；

（9）也就是說，如果層級不超過現(xiàn)有高度，只建立一條索引鏈，否則還要額外增加頭索引鏈的高度（腦補一下，后面舉例說明）；

Part III：將新建的索引節(jié)點（包含頭索引節(jié)點）與其它索引節(jié)點通過右指針連接在一起（補上right指針）

（1）從最高層級的頭索引節(jié)點開始，向右遍歷，找到目標(biāo)索引節(jié)點的位置；

（2）如果當(dāng)前層有目標(biāo)索引，則把目標(biāo)索引插入到這個位置，并把目標(biāo)索引前一個索引向下移一個層級；

（3）如果當(dāng)前層沒有目標(biāo)索引，則把目標(biāo)索引位置前一個索引向下移一個層級；

（4）同樣地，再向右遍歷，尋找新的層級中目標(biāo)索引的位置，回到第（2）步；

（5）依次循環(huán)找到所有層級目標(biāo)索引的位置并把它們插入到橫向的索引鏈表中；

總結(jié)起來，一共就是三大步：

（1）插入目標(biāo)節(jié)點到數(shù)據(jù)節(jié)點鏈表中；

（2）建立豎直的down鏈表；

（3）建立橫向的right鏈表；

添加元素舉例

假設(shè)初始鏈表是這樣：

假如，我們現(xiàn)在要插入一個元素9。

（1）尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點，在這里也就是找到了5這個數(shù)據(jù)節(jié)點；

（2）從5開始向后遍歷，找到目標(biāo)節(jié)點的位置，也就是在8和12之間；

（3）插入9這個元素，Part I 結(jié)束；

然后，計算其索引層級，假如是3，也就是level=3。

（1）建立豎直的down索引鏈表；

（2）超過了現(xiàn)有高度2，還要再增加head索引鏈的高度；

（3）至此，Part II 結(jié)束；

最后，把right指針補齊。

（1）從第3層的head往右找當(dāng)前層級目標(biāo)索引的位置；

（2）找到就把目標(biāo)索引和它前面索引的right指針連上，這里前一個正好是head；

（3）然后前一個索引向下移，這里就是head下移；

（4）再往右找目標(biāo)索引的位置；

（5）找到了就把right指針連上，這里前一個是3的索引；

（6）然后3的索引下移；

（7）再往右找目標(biāo)索引的位置；

（8）找到了就把right指針連上，這里前一個是5的索引；

（9）然后5下移，到底了，Part III 結(jié)束，整個插入過程結(jié)束；

是不是很簡單^^

刪除元素

刪除元素，就是把各層級中對應(yīng)的元素刪除即可，真的這么簡單嗎？來讓我們上代碼：

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

final V doRemove(Object key, Object value) {
    // key不為空
    if (key == null)
        throw new NullPointerException();
    Comparator<? super K> cmp = comparator;
    // 自旋
    outer: for (;;) {
        // 尋找目標(biāo)節(jié)點之前的最近的索引節(jié)點對應(yīng)的數(shù)據(jù)節(jié)點
        // 為了方便，這里叫b為當(dāng)前節(jié)點，n為下一個節(jié)點，f為下下個節(jié)點
        for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
            Object v; int c;
            // 整個鏈表都遍歷完了也沒找到目標(biāo)節(jié)點，退出外層循環(huán)
            if (n == null)
                break outer;
            // 下下個節(jié)點
            Node<K,V> f = n.next;
            // 再次檢查
            // 如果n不是b的下一個節(jié)點了
            // 說明有其它線程先一步修改了，從頭來過
            if (n != b.next)                    // inconsistent read
                break;
            // 如果下個節(jié)點的值奕為null了
            // 說明有其它線程標(biāo)記該元素為刪除狀態(tài)了
            if ((v = n.value) == null) {        // n is deleted
                // 協(xié)助刪除
                n.helpDelete(b, f);
                break;
            }
            // 如果b的值為空或者v等于n，說明b已被刪除
            // 這時候n就是marker節(jié)點，那b就是被刪除的那個
            if (b.value == null || v == n)      // b is deleted
                break;
            // 如果c<0，說明沒找到元素，退出外層循環(huán)
            if ((c = cpr(cmp, key, n.key)) < 0)
                break outer;
            // 如果c>0，說明還沒找到，繼續(xù)向右找
            if (c > 0) {
                // 當(dāng)前節(jié)點往后移
                b = n;
                // 下一個節(jié)點往后移
                n = f;
                continue;
            }
            // c=0，說明n就是要找的元素
            // 如果value不為空且不等于找到元素的value，不需要刪除，退出外層循環(huán)
            if (value != null && !value.equals(v))
                break outer;
            // 如果value為空，或者相等
            // 原子標(biāo)記n的value值為空
            if (!n.casValue(v, null))
                // 如果刪除失敗，說明其它線程先一步修改了，從頭來過
                break;

            // P.S.到了這里n的值肯定是設(shè)置成null了

            // 關(guān)鍵?。。。?            // 讓n的下一個節(jié)點指向一個market節(jié)點
            // 這個market節(jié)點的key為null，value為marker自己，next為n的下個節(jié)點f
            // 或者讓b的下一個節(jié)點指向下下個節(jié)點
            // 注意：這里是或者||，因為兩個CAS不能保證都成功，只能一個一個去嘗試
            // 這里有兩層意思：
            // 一是如果標(biāo)記market成功，再嘗試將b的下個節(jié)點指向下下個節(jié)點，如果第二步失敗了，進(jìn)入條件，如果成功了就不用進(jìn)入條件了
            // 二是如果標(biāo)記market失敗了，直接進(jìn)入條件
            if (!n.appendMarker(f) || !b.casNext(n, f))
                // 通過findNode()重試刪除（里面有個helpDelete()方法）
                findNode(key);                  // retry via findNode
            else {
                // 上面兩步操作都成功了，才會進(jìn)入這里，不太好理解，上面兩個條件都有非"!"操作
                // 說明節(jié)點已經(jīng)刪除了，通過findPredecessor()方法刪除索引節(jié)點
                // findPredecessor()里面有unlink()操作
                findPredecessor(key, cmp);      // clean index
                // 如果最高層頭索引節(jié)點沒有右節(jié)點，則跳表的高度降級
                if (head.right == null)
                    tryReduceLevel();
            }
            // 返回刪除的元素值
            @SuppressWarnings("unchecked") V vv = (V)v;
            return vv;
        }
    }
    return null;
}

（1）尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點（數(shù)據(jù)節(jié)點都是在最底層的鏈表上）；

（2）從這個數(shù)據(jù)節(jié)點開始往后遍歷，直到找到目標(biāo)節(jié)點的位置；

（3）如果這個位置沒有元素，直接返回null，表示沒有要刪除的元素；

（4）如果這個位置有元素，先通過n.casValue(v, null)原子更新把其value設(shè)置為null；

（5）通過n.appendMarker(f)在當(dāng)前元素后面添加一個marker元素標(biāo)記當(dāng)前元素是要刪除的元素；

（6）通過b.casNext(n, f)嘗試刪除元素；

（7）如果上面兩步中的任意一步失敗了都通過findNode(key)中的n.helpDelete(b, f)再去不斷嘗試刪除；

（8）如果上面兩步都成功了，再通過findPredecessor(key, cmp)中的q.unlink(r)刪除索引節(jié)點；

（9）如果head的right指針指向了null，則跳表高度降級；

刪除元素舉例

假如初始跳表如下圖所示，我們要刪除9這個元素。

（1）找到9這個數(shù)據(jù)節(jié)點；

（2）把9這個節(jié)點的value值設(shè)置為null；

（3）在9后面添加一個marker節(jié)點，標(biāo)記9已經(jīng)刪除了；

（4）讓8指向12；

（5）把索引節(jié)點與它前一個索引的right斷開聯(lián)系；

（6）跳表高度降級；

至于，為什么要有（2）（3）（4）這么多步驟呢，因為多線程下如果直接讓8指向12，可以其它線程先一步在9和12間插入了一個元素10呢，這時候就不對了。

所以這里搞了三步來保證多線程下操作的正確性。

如果第（2）步失敗了，則直接重試；

如果第（3）或（4）步失敗了，因為第（2）步是成功的，則通過helpDelete()不斷重試去刪除；

其實helpDelete()里面也是不斷地重試（3）和（4）；

只有這三步都正確完成了，才能說明這個元素徹底被刪除了。

這一塊結(jié)合上面圖中的紅綠藍(lán)色好好理解一下，一定要想在并發(fā)環(huán)境中會怎么樣。

查找元素

經(jīng)過上面的插入和刪除，查找元素就比較簡單了，直接上代碼：

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

private V doGet(Object key) {
    // key不為空
    if (key == null)
        throw new NullPointerException();
    Comparator<? super K> cmp = comparator;
    // 自旋
    outer: for (;;) {
        // 尋找目標(biāo)節(jié)點之前最近的索引對應(yīng)的數(shù)據(jù)節(jié)點
        // 為了方便，這里叫b為當(dāng)前節(jié)點，n為下個節(jié)點，f為下下個節(jié)點
        for (Node<K,V> b = findPredecessor(key, cmp), n = b.next;;) {
            Object v; int c;
            // 如果鏈表到頭還沒找到元素，則跳出外層循環(huán)
            if (n == null)
                break outer;
            // 下下個節(jié)點
            Node<K,V> f = n.next;
            // 如果不一致讀，從頭來過
            if (n != b.next)                // inconsistent read
                break;
            // 如果n的值為空，說明節(jié)點已被其它線程標(biāo)記為刪除
            if ((v = n.value) == null) {    // n is deleted
                // 協(xié)助刪除，再重試
                n.helpDelete(b, f);
                break;
            }
            // 如果b的值為空或者v等于n，說明b已被刪除
            // 這時候n就是marker節(jié)點，那b就是被刪除的那個
            if (b.value == null || v == n)  // b is deleted
                break;
            // 如果c==0，說明找到了元素，就返回元素值
            if ((c = cpr(cmp, key, n.key)) == 0) {
                @SuppressWarnings("unchecked") V vv = (V)v;
                return vv;
            }
            // 如果c<0，說明沒找到元素
            if (c < 0)
                break outer;
            // 如果c>0，說明還沒找到，繼續(xù)尋找
            // 當(dāng)前節(jié)點往后移
            b = n;
            // 下一個節(jié)點往后移
            n = f;
        }
    }
    return null;
}

（1）尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點（數(shù)據(jù)節(jié)點都是在最底層的鏈表上）；

（2）從這個數(shù)據(jù)節(jié)點開始往后遍歷，直到找到目標(biāo)節(jié)點的位置；

（3）如果這個位置沒有元素，直接返回null，表示沒有找到元素；

（4）如果這個位置有元素，返回元素的value值；

查找元素舉例

假如有如下圖所示這個跳表，我們要查找9這個元素，它走過的路徑是怎樣的呢？可能跟你相像的不一樣。。

（1）尋找目標(biāo)節(jié)點之前最近的一個索引對應(yīng)的數(shù)據(jù)節(jié)點，這里就是5；

（2）從5開始往后遍歷，經(jīng)過8，到9；

（3）找到了返回；

整個路徑如下圖所示：

是不是很操蛋？

為啥不從9的索引直接過來呢？

從我實際打斷點調(diào)試來看確實是按照上圖的路徑來走的。

我猜測可能是因為findPredecessor()這個方法是插入、刪除、查找元素多個方法共用的，在單鏈表中插入和刪除元素是需要記錄前一個元素的，而查找并不需要，這里為了兼容三者使得編碼相對簡單一點，所以就使用了同樣的邏輯，而沒有單獨對查找元素進(jìn)行優(yōu)化。

不過也可能是Doug Lea大神不小心寫了個bug，如果有人知道原因請告訴我。（公眾號后臺留言，新公眾號的文章下面不支持留言了，蛋疼）

彩蛋

為什么Redis選擇使用跳表而不是紅黑樹來實現(xiàn)有序集合？

請查看【拜托，面試別再問我跳表了！】這篇文章。

歡迎關(guān)注我的公眾號“彤哥讀源碼”，查看更多源碼系列文章, 與彤哥一起暢游源碼的海洋。