摘要：同時(shí)，也提供了一個(gè)基于的實(shí)現(xiàn)類，底層基于紅黑樹設(shè)計(jì)，是一種有序的。可以看成是并發(fā)版本的，但是和不同是，并不是基于紅黑樹實(shí)現(xiàn)的，其底層是一種類似跳表的結(jié)構(gòu)。上述所有構(gòu)造器都調(diào)用了方法方法將一些字段置初始化，然后將指針指向新創(chuàng)建的結(jié)點(diǎn)。
本文首發(fā)于一世流云專欄：https://segmentfault.com/blog...
一、ConcurrentSkipListMap簡(jiǎn)介
類繼承結(jié)構(gòu)
在正式講ConcurrentSkipListMap之前，我們先來看下ConcurrentSkipListMap的類繼承圖：
我們知道，一般的Map都是無序的，也就是只能通過鍵的hash值進(jìn)行定位。JDK為了實(shí)現(xiàn)有序的Map，提供了一個(gè)SortedMap接口，SortedMap提供了一些根據(jù)鍵范圍進(jìn)行查找的功能，比如返回整個(gè)Map中 key最小/大的鍵、返回某個(gè)范圍內(nèi)的子Map視圖等等。
為了進(jìn)一步對(duì)有序Map進(jìn)行增強(qiáng)，JDK又引入了NavigableMap接口，該接口進(jìn)一步擴(kuò)展了SortedMap的功能，提供了根據(jù)指定Key返回最接近項(xiàng)、按升序/降序返回所有鍵的視圖等功能。
同時(shí)，也提供了一個(gè)基于NavigableMap的實(shí)現(xiàn)類——TreeMap，TreeMap底層基于紅黑樹設(shè)計(jì)，是一種有序的Map。關(guān)于TreeMap和NavigableMap，本文不作贅述，讀者可以查看Oracle的官方文檔：https://docs.oracle.com/javas...。
ConcurrentSkipListMap的由來
JDK1.6時(shí)，為了對(duì)高并發(fā)環(huán)境下的有序Map提供更好的支持，J.U.C新增了一個(gè)ConcurrentNavigableMap接口，ConcurrentNavigableMap很簡(jiǎn)單，它同時(shí)實(shí)現(xiàn)了NavigableMap和ConcurrentMap接口：
ConcurrentNavigableMap接口提供的功能也和NavigableMap幾乎完全一致，很多方法僅僅是返回的類型不同：
J.U.C提供了基于ConcurrentNavigableMap接口的一個(gè)實(shí)現(xiàn)——ConcurrentSkipListMap。ConcurrentSkipListMap可以看成是并發(fā)版本的TreeMap，但是和TreeMap不同是，ConcurrentSkipListMap并不是基于紅黑樹實(shí)現(xiàn)的，其底層是一種類似跳表（Skip List）的結(jié)構(gòu)。
二、Skip List簡(jiǎn)介
什么是Skip List
Skip List（以下簡(jiǎn)稱跳表），是一種類似鏈表的數(shù)據(jù)結(jié)構(gòu)，其查詢/插入/刪除的時(shí)間復(fù)雜度都是O(logn)。
我們知道，通常意義上的鏈表是不能支持隨機(jī)訪問的（通過索引快速定位），其查找的時(shí)間復(fù)雜度是O(n)，而數(shù)組這一可支持隨機(jī)訪問的數(shù)據(jù)結(jié)構(gòu)，雖然查找很快，但是插入/刪除元素卻需要移動(dòng)插入點(diǎn)后的所有元素，時(shí)間復(fù)雜度為O(n)。
為了解決這一問題，引入了樹結(jié)構(gòu)，樹的增刪改查效率比較平均，一棵平衡二叉樹（AVL）的增刪改查效率一般為O(logn)，比如工業(yè)上常用紅黑樹作為AVL的一種實(shí)現(xiàn)。
但是，AVL的實(shí)現(xiàn)一般都比較復(fù)雜，插入/刪除元素可能涉及對(duì)整個(gè)樹結(jié)構(gòu)的修改，特別是并發(fā)環(huán)境下，通常需要全局鎖來保證AVL的線程安全，于是又出現(xiàn)了一種類似鏈表的數(shù)據(jù)結(jié)構(gòu)——跳表。
Skip List示例
在講Skip List之前，我們先來看下傳統(tǒng)的單鏈表：
上圖的單鏈表中（省去了結(jié)點(diǎn)之間的鏈接），當(dāng)想查找7、15、46這三個(gè)元素時(shí)，必須從頭指針head開始，遍歷整個(gè)單鏈表，其查找復(fù)雜度很低，為O(n)。
來看下Skip List的數(shù)據(jù)結(jié)構(gòu)是什么樣的：
上圖是Skip List一種可能的結(jié)構(gòu)，它分了2層，假設(shè)我們要查找“15”這個(gè)元素，那么整個(gè)步驟如下：
從頭指針head開始，找到第一個(gè)結(jié)點(diǎn)的最上層，發(fā)現(xiàn)其指向的下個(gè)結(jié)點(diǎn)值為8，小于15，則直接從1結(jié)點(diǎn)跳到8結(jié)點(diǎn)。
8結(jié)點(diǎn)最上層指向的下一結(jié)點(diǎn)值為18，大于15，則從8結(jié)點(diǎn)的下一層開始查找。
從8結(jié)點(diǎn)的最下層一直向后查找，依次經(jīng)過10、13，最后找到15結(jié)點(diǎn)。
上述整個(gè)查找路徑如下圖標(biāo)黃部分所示：
同理，如果要查找“46”這個(gè)元素，則整個(gè)查找路徑如下圖標(biāo)黃部分所示：
上面就是跳躍表的基本思想了，每個(gè)結(jié)點(diǎn)不僅僅只包含指向下一個(gè)結(jié)點(diǎn)的指針，可能還包含很多個(gè)其它指向后續(xù)結(jié)點(diǎn)的指針。并且，一個(gè)結(jié)點(diǎn)本身可以看成是一個(gè)鏈表（自上向下鏈接）。這樣就可以跳過一些不必要的結(jié)點(diǎn)，從而加快查找、刪除等操作，這其實(shí)是一種“空間換時(shí)間”的算法設(shè)計(jì)思想。
那么一個(gè)結(jié)點(diǎn)可以包含多少層呢？ 比如，Skip List也可能是下面這種包含3層的結(jié)構(gòu)(在一個(gè)3層Skip List中查找元素“46”)：
層數(shù)是根據(jù)一種隨機(jī)算法得到的，為了不讓層數(shù)過大，還會(huì)有一個(gè)最大層數(shù)MAX_LEVEL限制，隨機(jī)算法生成的層數(shù)不得大于該值。后面講ConcurrentSkipListMap時(shí)，我們會(huì)具體分析。
以上就是Skip List的基本思想了，總結(jié)起來，有以下幾點(diǎn)：
跳表由很多層組成；
每一層都是一個(gè)有序鏈表；
對(duì)于每一層的任意結(jié)點(diǎn)，不僅有指向下一個(gè)結(jié)點(diǎn)的指針，也有指向其下一層的指針。
三、ConcurrentSkipListMap的內(nèi)部結(jié)構(gòu)
介紹完了跳表，再來看ConcurrentSkipListMap的內(nèi)部結(jié)構(gòu)就容易得多了：
ConcurrentSkipListMap內(nèi)部一共定義了3種不同類型的結(jié)點(diǎn)，元素的增刪改查都從最上層的head指針指向的結(jié)點(diǎn)開始：
public class ConcurrentSkipListMap2 extends AbstractMap
implements ConcurrentNavigableMap, Cloneable, Serializable {
/**
* 最底層鏈表的頭指針BASE_HEADER
*/
private static final Object BASE_HEADER = new Object();
/**
* 最上層鏈表的頭指針head
*/
private transient volatile HeadIndex head;
/* ---------------- 普通結(jié)點(diǎn)Node定義 -------------- */
static final class Node {
final K key;
volatile Object value;
volatile Node next;
// ...
}
/* ---------------- 索引結(jié)點(diǎn)Index定義 -------------- */
static class Index {
final Node node; // node指向最底層鏈表的Node結(jié)點(diǎn)
final Index down; // down指向下層Index結(jié)點(diǎn)
volatile Index right; // right指向右邊的Index結(jié)點(diǎn)
// ...
}
/* ---------------- 頭索引結(jié)點(diǎn)HeadIndex -------------- */
static final class HeadIndex extends Index {
final int level; // 層級(jí)
// ...
}
}
我們來看下這3類結(jié)點(diǎn)的具體定義。
結(jié)點(diǎn)定義
普通結(jié)點(diǎn)：Node
普通結(jié)點(diǎn)——Node，也就是ConcurrentSkipListMap最底層鏈表中的結(jié)點(diǎn)，保存著實(shí)際的鍵值對(duì)，如果多帶帶看底層鏈，其實(shí)就是一個(gè)按照Key有序排列的單鏈表：
static final class Node {
final K key;
volatile Object value;
volatile Node next;
/**
* 正常結(jié)點(diǎn).
*/
Node(K key, Object value, Node next) {
this.key = key;
this.value = value;
this.next = next;
}
/**
* 標(biāo)記結(jié)點(diǎn).
*/
Node(Node next) {
this.key = null;
this.value = this;
this.next = next;
}
/**
* CAS更新結(jié)點(diǎn)的value
*/
boolean casValue(Object cmp, Object val) {
return UNSAFE.compareAndSwapObject(this, valueOffset, cmp, val);
}
/**
* CAS更新結(jié)點(diǎn)的next
*/
boolean casNext(Node cmp, Node val) {
return UNSAFE.compareAndSwapObject(this, nextOffset, cmp, val);
}
/**
* 判斷當(dāng)前結(jié)點(diǎn)是否為[標(biāo)記結(jié)點(diǎn)]
*/
boolean isMarker() {
return value == this;
}
/**
* 判斷當(dāng)前結(jié)點(diǎn)是否是最底層鏈表的頭結(jié)點(diǎn)
*/
boolean isBaseHeader() {
return value == BASE_HEADER;
}
/**
* 在當(dāng)前結(jié)點(diǎn)后面插入一個(gè)標(biāo)記結(jié)點(diǎn).
*
* @param f 當(dāng)前結(jié)點(diǎn)的后繼結(jié)點(diǎn)
* @return true 插入成功
*/
boolean appendMarker(Node f) {
return casNext(f, new Node(f));
}
/**
* 輔助刪除結(jié)點(diǎn)方法.
*
* @param b 當(dāng)前結(jié)點(diǎn)的前驅(qū)結(jié)點(diǎn)
* @param f 當(dāng)前結(jié)點(diǎn)的后繼結(jié)點(diǎn)
*/
void helpDelete(Node b, Node f) {
/*
* 重新檢查一遍結(jié)點(diǎn)位置
* 確保b和f分別為當(dāng)前結(jié)點(diǎn)的前驅(qū)/后繼
*/
if (f == next && this == b.next) {
if (f == null || f.value != f) // f為null或非標(biāo)記結(jié)點(diǎn)
casNext(f, new Node(f));
else // 刪除當(dāng)前結(jié)點(diǎn)
b.casNext(this, f.next);
}
}
/**
* 返回結(jié)點(diǎn)的value值.
*
* @return 標(biāo)記結(jié)點(diǎn)或最底層頭結(jié)點(diǎn)，直接返回null
*/
V getValidValue() {
Object v = value;
if (v == this || v == BASE_HEADER) // 標(biāo)記結(jié)點(diǎn)或最底層頭結(jié)點(diǎn)，直接返回null
return null;
V vv = (V) v;
return vv;
}
/**
* 返回當(dāng)前結(jié)點(diǎn)的一個(gè)Immutable快照.
*/
AbstractMap.SimpleImmutableEntry createSnapshot() {
Object v = value;
if (v == null || v == this || v == BASE_HEADER)
return null;
V vv = (V) v;
return new AbstractMap.SimpleImmutableEntry(key, vv);
}
// UNSAFE mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long valueOffset;
private static final long nextOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class k = Node.class;
valueOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("value"));
nextOffset = UNSAFE.objectFieldOffset
(k.getDeclaredField("next"));
} catch (Exception e) {
throw new Error(e);
}
}
}
索引結(jié)點(diǎn)：Index
Index結(jié)點(diǎn)是除底層鏈外，其余各層鏈表中的非頭結(jié)點(diǎn)（見示意圖中的藍(lán)色結(jié)點(diǎn)）。每個(gè)Index結(jié)點(diǎn)包含3個(gè)指針：down、right、node。
down和right指針分別指向下層結(jié)點(diǎn)和后繼結(jié)點(diǎn)，node指針指向其最底部的node結(jié)點(diǎn)。
static class Index {
final Node node; // node指向最底層鏈表的Node結(jié)點(diǎn)
final Index down; // down指向下層Index結(jié)點(diǎn)
volatile Index right; // right指向右邊的Index結(jié)點(diǎn)
Index(Node node, Index down, Index right) {
this.node = node;
this.down = down;
this.right = right;
}
/**
* CAS更新右邊的Index結(jié)點(diǎn)
*
* @param cmp 當(dāng)前結(jié)點(diǎn)的右結(jié)點(diǎn)
* @param val 希望更新的結(jié)點(diǎn)
*/
final boolean casRight(Index cmp, Index val) {
return UNSAFE.compareAndSwapObject(this, rightOffset, cmp, val);
}
/**
* 判斷Node結(jié)點(diǎn)是否已經(jīng)刪除.
*/
final boolean indexesDeletedNode() {
return node.value == null;
}
/**
* CAS插入一個(gè)右邊結(jié)點(diǎn)newSucc.
*
* @param succ 當(dāng)前的后繼結(jié)點(diǎn)
* @param newSucc 新的后繼結(jié)點(diǎn)
*/
final boolean link(Index succ, Index newSucc) {
Node n = node;
newSucc.right = succ;
return n.value != null && casRight(succ, newSucc);
}
/**
* 跳過當(dāng)前結(jié)點(diǎn)的后繼結(jié)點(diǎn).
*
* @param succ 當(dāng)前的后繼結(jié)點(diǎn)
*/
final boolean unlink(Index succ) {
return node.value != null && casRight(succ, succ.right);
}
// Unsafe mechanics
private static final sun.misc.Unsafe UNSAFE;
private static final long rightOffset;
static {
try {
UNSAFE = sun.misc.Unsafe.getUnsafe();
Class k = Index.class;
rightOffset = UNSAFE.objectFieldOffset(k.getDeclaredField("right"));
} catch (Exception e) {
throw new Error(e);
}
}
}
頭索引結(jié)點(diǎn)：HeadIndex
HeadIndex結(jié)點(diǎn)是各層鏈表的頭結(jié)點(diǎn)，它是Index類的子類，唯一的區(qū)別是增加了一個(gè)level字段，用于表示當(dāng)前鏈表的級(jí)別，越往上層，level值越大。
static final class HeadIndex extends Index {
final int level; // 層級(jí)
HeadIndex(Node node, Index down, Index right, int level) {
super(node, down, right);
this.level = level;
}
}
構(gòu)造器定義和初始化
ConcurrentSkipListMap一共定義了4種構(gòu)造器：
空構(gòu)造器
/**
* 構(gòu)造一個(gè)新的空Map.
*/
public ConcurrentSkipListMap() {
this.comparator = null;
initialize();
}
指定比較器的構(gòu)造器
/**
* 構(gòu)造一個(gè)新的空Map.
* 并指定比較器.
*/
public ConcurrentSkipListMap(Comparator comparator) {
this.comparator = comparator;
initialize();
}
從給定Map構(gòu)建的構(gòu)造器
/**
* 從已給定的Map構(gòu)造一個(gè)新Map.
*/
public ConcurrentSkipListMap(Map m) {
this.comparator = null;
initialize();
putAll(m);
}
從給定SortedMap構(gòu)建的構(gòu)造器
/**
* 從已給定的SortedMap構(gòu)造一個(gè)新Map.
* 并且Key的順序與原來保持一致.
*/
public ConcurrentSkipListMap(SortedMap m) {
this.comparator = m.comparator();
initialize();
buildFromSorted(m);
}
注：ConcurrentSkipListMap會(huì)基于比較器——Comparator ，來進(jìn)行鍵Key的比較，如果構(gòu)造時(shí)未指定Comparator ，那么就會(huì)按照Key的自然順序進(jìn)行比較，所謂Key的自然順序是指key實(shí)現(xiàn)Comparable接口。
上述所有構(gòu)造器都調(diào)用了initialize方法：
private void initialize() {
keySet = null;
entrySet = null;
values = null;
descendingMap = null;
head = new HeadIndex(new Node(null, BASE_HEADER, null),null, null, 1);
}
initialize方法將一些字段置初始化null，然后將head指針指向新創(chuàng)建的HeadIndex結(jié)點(diǎn)。初始化完成后，ConcurrentSkipListMap的結(jié)構(gòu)如下：
其中，head和BASE_HEADER都是ConcurrentSkipListMap的字段：
/**
* 最底層鏈表的頭指針BASE_HEADER
*/
private static final Object BASE_HEADER = new Object();
/**
* 最上層鏈表的頭指針head
*/
private transient volatile HeadIndex head;
四、ConcurrentSkipListMap的核心操作
put操作
put操作本身很簡(jiǎn)單，需要注意的是ConcurrentSkipListMap在插入鍵值對(duì)時(shí)，Key和Value都不能為null：
/**
* 插入鍵值對(duì).
*
* @param key 鍵
* @param value 值
* @return 如果key存在，返回舊value值；否則返回null
*/
public V put(K key, V value) {
if (value == null) // ConcurrentSkipListMap的Value不能為null
throw new NullPointerException();
return doPut(key, value, false);
}
上述方法內(nèi)部調(diào)用了doPut來做實(shí)際的插入操作：
/**
* 插入鍵值對(duì).
*
* @param onlyIfAbsent true: 僅當(dāng)Key不存在時(shí)才進(jìn)行插入
*/
private V doPut(K key, V value, boolean onlyIfAbsent) {
Node z; // z指向待添加的Node結(jié)點(diǎn)
if (key == null) // ConcurrentSkipListMap的Key不能為null
throw new NullPointerException();
Comparator cmp = comparator;
outer:
for (; ; ) {
// b是“是小于且最接近給定key”的Node結(jié)點(diǎn)(或底層鏈表頭結(jié)點(diǎn))
for (Node b = findPredecessor(key, cmp), n = b.next; ; ) {
if (n != null) { // b存在后驅(qū)結(jié)點(diǎn)： b -> n -> f
Object v;
int c;
Node f = n.next; // f指向b的后驅(qū)的后驅(qū)
if (n != b.next) // 存在并發(fā)修改,放棄并重試
break;
if ((v = n.value) == null) { // n為標(biāo)記刪除結(jié)點(diǎn)
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n) // b為標(biāo)記刪除結(jié)點(diǎn)
break;
if ((c = cpr(cmp, key, n.key)) > 0) { // 向后遍歷,找到第一個(gè)大于key的結(jié)點(diǎn)
b = n;
n = f;
continue;
}
if (c == 0) { // 存在Key相同的結(jié)點(diǎn)
if (onlyIfAbsent || n.casValue(v, value)) {
V vv = (V) v;
return vv;
}
break; // CAS更新失敗,則重試
}
}
z = new Node(key, value, n);
if (!b.casNext(n, z)) // 嘗試插入z結(jié)點(diǎn)： b -> z -> n
break; // CAS插入失敗，則重試
break outer; // 跳出最外層循環(huán)
}
}
int rnd = ThreadLocalRandom.nextSecondarySeed(); // 生成一個(gè)隨機(jī)數(shù)種子
if ((rnd & 0x80000001) == 0) { // 為true表示需要增加層級(jí)
/**
* 以下方法用于創(chuàng)建新層級(jí)
*/
int level = 1, max;
while (((rnd >>>= 1) & 1) != 0) // level表示新的層級(jí),通過下面這個(gè)while循環(huán)可以確認(rèn)新的層級(jí)數(shù)
++level;
Index idx = null;
HeadIndex h = head;
if (level <= (max = h.level)) { // CASE1: 新層級(jí)level沒有超過最大層級(jí)head.level（head指針指向最高層）
// 以“頭插法”創(chuàng)建level個(gè)Index結(jié)點(diǎn),idx最終指向最高層的Index結(jié)點(diǎn)
for (int i = 1; i <= level; ++i)
idx = new Index(z, idx, null);
}
else { // CASE2: 新層級(jí)level超過了最大層級(jí)head.level
level = max + 1; // 重置level為最大層級(jí)+1
// 生成一個(gè)Index結(jié)點(diǎn)數(shù)組,idxs[0]不會(huì)使用
Index[] idxs = (Index[]) new Index[level + 1];
for (int i = 1; i <= level; ++i)
idxs[i] = idx = new Index(z, idx, null);
// 生成新的HeadIndex結(jié)點(diǎn)
for (; ; ) {
h = head;
int oldLevel = h.level; // 原最大層級(jí)
if (level <= oldLevel)
break;
HeadIndex newh = h;
Node oldbase = h.node; // oldbase指向最底層鏈表的頭結(jié)點(diǎn)
for (int j = oldLevel + 1; j <= level; ++j)
newh = new HeadIndex(oldbase, newh, idxs[j], j);
if (casHead(h, newh)) {
h = newh;
idx = idxs[level = oldLevel];
break;
}
}
}
/**
* 以下方法用于鏈接新層級(jí)的各個(gè)HeadIndex和Index結(jié)點(diǎn)
*/
splice:
for (int insertionLevel = level; ; ) { // 此時(shí)level為oldLevel，即原最大層級(jí)
int j = h.level;
for (Index q = h, r = q.right, t = idx; ; ) {
if (q == null || t == null)
break splice;
if (r != null) {
Node n = r.node;
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)) // 在q和r之間插入t，即從 q -> r 變成 q -> t -> r
break;
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方法，而是看下其內(nèi)部的findPredecessor方法，findPredecessor用于查找“小于且最接近給定key”的Node結(jié)點(diǎn)，并且這個(gè)Node結(jié)點(diǎn)必須有上層結(jié)點(diǎn)：
/**
* 返回“小于且最接近給定key”的數(shù)據(jù)結(jié)點(diǎn).
* 如果不存在這樣的數(shù)據(jù)結(jié)點(diǎn)，則返回底層鏈表的頭結(jié)點(diǎn).
*
* @param key 待查找的鍵
*/
private Node findPredecessor(Object key, Comparator cmp) {
if (key == null)
throw new NullPointerException();
/**
* 從最上層開始，往右下方向查找
*/
for (; ; ) {
for (Index q = head, r = q.right, d; ; ) { // 從最頂層的head結(jié)點(diǎn)開始查找
if (r != null) { // 存在右結(jié)點(diǎn)
Node n = r.node;
K k = n.key;
if (n.value == null) { // 處理結(jié)點(diǎn)”懶刪除“的情況
if (!q.unlink(r))
break;
r = q.right;
continue;
}
if (cpr(cmp, key, k) > 0) { // key大于k,繼續(xù)向右查找
q = r;
r = r.right;
continue;
}
}
//已經(jīng)到了level1的層
if ((d = q.down) == null) // 不存在下結(jié)點(diǎn)，說明q已經(jīng)是level1鏈表中的結(jié)點(diǎn)了
return q.node; // 直接返回對(duì)應(yīng)的Node結(jié)點(diǎn)
// 轉(zhuǎn)到下一層，繼續(xù)查找(level-1層)
q = d;
r = d.right;
}
}
}
看代碼不太直觀，我們還是看下面這個(gè)圖：
上圖中，假設(shè)要查找的Key為72，則步驟如下：
從最上方head指向的結(jié)點(diǎn)開始，比較①號(hào)標(biāo)紅的Index結(jié)點(diǎn)的key值，發(fā)現(xiàn)3小于72，則繼續(xù)向右；
比較②號(hào)標(biāo)紅的Index結(jié)點(diǎn)的key值，發(fā)現(xiàn)62小于72，則繼續(xù)向右
由于此時(shí)右邊是null，則轉(zhuǎn)而向下，一直到⑥號(hào)標(biāo)紅結(jié)點(diǎn)；
由于⑥號(hào)標(biāo)紅結(jié)點(diǎn)的down字段為空（不能再往下了，已經(jīng)是level1最低層了），則直接返回它的node字段指向的結(jié)點(diǎn)，即⑧號(hào)結(jié)點(diǎn)。
注意：如果我們要查找key為59的Node結(jié)點(diǎn)，返回的不是Key為45的結(jié)點(diǎn)，而是key為23的結(jié)點(diǎn)。讀者可以自己在紙上比劃下。
回到doPut方法，假設(shè)現(xiàn)在待插入的Key為3，則當(dāng)執(zhí)行完下面這段代碼后，ConcurrentSkipListMap的結(jié)構(gòu)如下：
/**
* 插入鍵值對(duì).
*
* @param onlyIfAbsent true: 僅當(dāng)Key不存在時(shí)才進(jìn)行插入
*/
private V doPut(K key, V value, boolean onlyIfAbsent) {
Node z; // z指向待添加的Node結(jié)點(diǎn)
if (key == null) // ConcurrentSkipListMap的Key不能為null
throw new NullPointerException();
Comparator cmp = comparator;
outer:
for (; ; ) {
// b是“是小于且最接近給定key”的Node結(jié)點(diǎn)(或底層鏈表頭結(jié)點(diǎn))
for (Node b = findPredecessor(key, cmp), n = b.next; ; ) {
if (n != null) { // b存在后驅(qū)結(jié)點(diǎn)： b -> n -> f
Object v;
int c;
Node f = n.next; // f指向b的后驅(qū)的后驅(qū)
if (n != b.next) // 存在并發(fā)修改,放棄并重試
break;
if ((v = n.value) == null) { // n為標(biāo)記刪除結(jié)點(diǎn)
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n) // b為標(biāo)記刪除結(jié)點(diǎn)
break;
if ((c = cpr(cmp, key, n.key)) > 0) { // 向后遍歷,找到第一個(gè)大于key的結(jié)點(diǎn)
b = n;
n = f;
continue;
}
if (c == 0) { // 存在Key相同的結(jié)點(diǎn)
if (onlyIfAbsent || n.casValue(v, value)) {
V vv = (V) v;
return vv;
}
break; // CAS更新失敗,則重試
}
}
z = new Node(key, value, n);
if (!b.casNext(n, z)) // 嘗試插入z結(jié)點(diǎn)： b -> z -> n
break; // CAS插入失敗，則重試
break outer; // 跳出最外層循環(huán)
}
// ...
}
}
上面是doPut中的第一個(gè)循環(huán)，作用就是找到底層鏈表的插入點(diǎn)，然后插入結(jié)點(diǎn)（在查找過程中可能會(huì)刪除一些已標(biāo)記的刪除結(jié)點(diǎn)）。
插入完成后，doPut方法并沒結(jié)束，我們之前說過ConcurrentSkipListMap的分層數(shù)是通過一個(gè)隨機(jī)數(shù)生成算法來確定，doPut的后半段，就是這個(gè)作用：判斷是否需要增加層級(jí)，如果需要就在各層級(jí)中插入對(duì)應(yīng)的Index結(jié)點(diǎn)。
/**
* 插入鍵值對(duì).
*
* @param onlyIfAbsent true: 僅當(dāng)Key不存在時(shí)才進(jìn)行插入
*/
private V doPut(K key, V value, boolean onlyIfAbsent) {
// ...
int rnd = ThreadLocalRandom.nextSecondarySeed(); // 生成一個(gè)隨機(jī)數(shù)種子
if ((rnd & 0x80000001) == 0) { // 為true表示需要增加層級(jí)
/**
* 以下方法用于創(chuàng)建新層級(jí)
*/
int level = 1, max;
while (((rnd >>>= 1) & 1) != 0) // level表示新的層級(jí),通過下面這個(gè)while循環(huán)可以確認(rèn)新的層級(jí)數(shù)
++level;
Index idx = null;
HeadIndex h = head;
if (level <= (max = h.level)) { // CASE1: 新層級(jí)level沒有超過最大層級(jí)head.level（head指針指向最高層）
// 以“頭插法”創(chuàng)建level個(gè)Index結(jié)點(diǎn),idx最終指向最高層的Index結(jié)點(diǎn)
for (int i = 1; i <= level; ++i)
idx = new Index(z, idx, null);
}
else { // CASE2: 新層級(jí)level超過了最大層級(jí)head.level
level = max + 1; // 重置level為最大層級(jí)+1
// 生成一個(gè)Index結(jié)點(diǎn)數(shù)組,idxs[0]不會(huì)使用
Index[] idxs = (Index[]) new Index[level + 1];
for (int i = 1; i <= level; ++i)
idxs[i] = idx = new Index(z, idx, null);
// 生成新的HeadIndex結(jié)點(diǎn)
for (; ; ) {
h = head;
int oldLevel = h.level; // 原最大層級(jí)
if (level <= oldLevel)
break;
HeadIndex newh = h;
Node oldbase = h.node; // oldbase指向最底層鏈表的頭結(jié)點(diǎn)
for (int j = oldLevel + 1; j <= level; ++j)
newh = new HeadIndex(oldbase, newh, idxs[j], j);
if (casHead(h, newh)) {
h = newh;
idx = idxs[level = oldLevel];
break;
}
}
}
/**
* 以下方法用于鏈接新層級(jí)的各個(gè)HeadIndex和Index結(jié)點(diǎn)
*/
splice:
for (int insertionLevel = level; ; ) { // 此時(shí)level為oldLevel，即原最大層級(jí)
int j = h.level;
for (Index q = h, r = q.right, t = idx; ; ) {
if (q == null || t == null)
break splice;
if (r != null) {
Node n = r.node;
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)) // 在q和r之間插入t，即從 q -> r 變成 q -> t -> r
break;
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;
}
最終ConcurrentSkipListMap的結(jié)構(gòu)如下所示：
remove操作
ConcurrentSkipListMap在刪除鍵值對(duì)時(shí)，不會(huì)立即執(zhí)行刪除，而是通過引入“標(biāo)記結(jié)點(diǎn)”，以“懶刪除”的方式進(jìn)行，以提高并發(fā)效率。
public V remove(Object key) {
return doRemove(key, null);
}
remove方法很簡(jiǎn)單，內(nèi)部調(diào)用了doRemove方法：
final V doRemove(Object key, Object value) {
if (key == null)
throw new NullPointerException();
Comparator cmp = comparator;
outer:
for (; ; ) {
// b指向“小于且最接近給定key”的Node結(jié)點(diǎn)(或底層鏈表頭結(jié)點(diǎn))
for (Node b = findPredecessor(key, cmp), n = b.next; ; ) { // b -> n
Object v;
int c;
if (n == null)
break outer;
Node f = n.next; // b -> n -> f
if (n != b.next) // 一致性判斷
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;
}
// 此時(shí)n指向查到的結(jié)點(diǎn)
if (value != null && !value.equals(v))
break outer;
if (!n.casValue(v, null)) // 更新查找到的結(jié)點(diǎn)的value為null
break;
// 在n和f之間添加標(biāo)記結(jié)點(diǎn)，并將b直接指向f
if (!n.appendMarker(f) || !b.casNext(n, f)) // n -> marker -> f
findNode(key); // retry via findNode
else {
findPredecessor(key, cmp); // 刪除Index結(jié)點(diǎn)
if (head.right == null) // 減少層級(jí)
tryReduceLevel();
}
V vv = (V) v;
return vv;
}
}
return null;
}
還是通過示例來理解上述代碼，假設(shè)現(xiàn)在要?jiǎng)h除Key==23的結(jié)點(diǎn)，刪除前ConcurrentSkipListMap的結(jié)構(gòu)如下：
doRemove方法首先會(huì)找到待刪除的結(jié)點(diǎn)，在它和后繼結(jié)點(diǎn)之間插入一個(gè)value為null的標(biāo)記結(jié)點(diǎn)（如下圖中的綠色結(jié)點(diǎn)），然后改變其前驅(qū)結(jié)點(diǎn)的指向：
最后，doRemove會(huì)重新調(diào)用一遍findPredecessor方法，解除被刪除結(jié)點(diǎn)上的Index結(jié)點(diǎn)之間的引用：
這樣Key==23的結(jié)點(diǎn)其實(shí)就被孤立，再后續(xù)查找或插入過程中，會(huì)被完全清除或被GC回收。
get操作
最后，我們來看下ConcurrentSkipListMap的查找操作——get方法。
public V get(Object key) {
return doGet(key);
}
內(nèi)部調(diào)用了doGet方法：
private V doGet(Object key) {
if (key == null)
throw new NullPointerException();
Comparator cmp = comparator;
outer:
for (; ; ) {
// b指向“小于且最接近給定key”的Node結(jié)點(diǎn)(或底層鏈表頭結(jié)點(diǎn))
for (Node b = findPredecessor(key, cmp), n = b.next; ; ) {
Object v;
int c;
if (n == null)
break outer;
Node f = n.next; // b -> n -> f
if (n != b.next)
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) {
V vv = (V) v;
return vv;
}
if (c < 0)
break outer;
b = n;
n = f;
}
}
return null;
}
doGet方法非常簡(jiǎn)單：
首先找到“小于且最接近給定key”的Node結(jié)點(diǎn)，然后用了三個(gè)指針：b -> n -> f，
n用于定位最終查找的Key，然后順著鏈表一步步向下查，比如查找KEY==45，則最終三個(gè)指針的位置如下：
