Java多線程（四）之ConcurrentSkipListMap深刻分析

1、前言 

concurrentHashMap與ConcurrentSkipListMap性能測試

在4線程1.6萬數據的條件下，ConcurrentHashMap 存取速度是ConcurrentSkipListMap 的4倍左右。

但ConcurrentSkipListMap有幾個ConcurrentHashMap 不能比擬的優勢：

一、ConcurrentSkipListMap 的key是有序的。

二、ConcurrentSkipListMap 支持更高的併發。ConcurrentSkipListMap 的存取時間是log（N），和線程數幾乎無關。也就是說在數據量必定的狀況下，併發的線程越多，ConcurrentSkipListMap越能體現出他的優點。 

2、使用建議

在非多線程的狀況下，應當儘可能使用TreeMap。此外對於併發性相對較低的並行程序可使用Collections.synchronizedSortedMap將TreeMap進行包裝，也能夠提供較好的效率。對於高併發程序，應當使用ConcurrentSkipListMap，可以提供更高的併發度。

因此在多線程程序中，若是須要對Map的鍵值進行排序時，請儘可能使用ConcurrentSkipListMap，可能獲得更好的併發度。
注意，調用ConcurrentSkipListMap的size時，因爲多個線程能夠同時對映射表進行操做，因此映射表須要遍歷整個鏈表才能返回元素個數，這個操做是個O(log(n))的操做。

2、什麼是SkipList

Skip list(跳錶）是一種能夠代替平衡樹的數據結構，默認是按照Key值升序的。Skip list讓已排序的數據分佈在多層鏈表中，以0-1隨機數決定一個數據的向上攀升與否，經過「空間來換取時間」的一個算法，在每一個節點中增長了向前的指針，在插入、刪除、查找時能夠忽略一些不可能涉及到的結點，從而提升了效率。

從機率上保持數據結構的平衡比顯示的保持數據結構平衡要簡單的多。對於大多數應用，用Skip list要比用樹算法相對簡單。因爲Skip list比較簡單，實現起來會比較容易，雖然和平衡樹有着相同的時間複雜度(O(logn)),可是skip list的常數項會相對小不少。Skip list在空間上也比較節省。一個節點平均只須要1.333個指針（甚至更少）。
                
圖1-1 Skip list結構圖（以7,14,21,32,37,71,85序列爲例）

Skip list的性質

(1) 由不少層結構組成，level是經過必定的機率隨機產生的。
(2) 每一層都是一個有序的鏈表，默認是升序，也能夠根據建立映射時所提供的Comparator進行排序，具體取決於使用的構造方法。
(3) 最底層(Level 1)的鏈表包含全部元素。
(4) 若是一個元素出如今Level i 的鏈表中，則它在Level i 之下的鏈表也都會出現。
(5) 每一個節點包含兩個指針，一個指向同一鏈表中的下一個元素，一個指向下面一層的元素。

3、什麼是ConcurrentSkipListMap

ConcurrentSkipListMap提供了一種線程安全的併發訪問的排序映射表。內部是SkipList（跳錶）結構實現，在理論上可以在O(log(n))時間內完成查找、插入、刪除操做。
       注意，調用ConcurrentSkipListMap的size時，因爲多個線程能夠同時對映射表進行操做，因此映射表須要遍歷整個鏈表才能返回元素個數，這個操做是個O(log(n))的操做。

 ConcurrentSkipListMap存儲結構

ConcurrentSkipListMap存儲結構圖

 

跳躍表（SkipList）：（如上圖所示）
1.多條鏈構成，是關鍵字升序排列的數據結構；
2.包含多個級別，一個head引用指向最高的級別，最低（底部）的級別，包含全部的key；
3.每個級別都是其更低級別的子集，而且是有序的；
4.若是關鍵字 key在 級別level=i中出現，則，level<=i的鏈表中都會包含該關鍵字key；

------------------------

ConcurrentSkipListMap主要用到了Node和Index兩種節點的存儲方式，經過volatile關鍵字實現了併發的操做

  

[java]  view plain copy

1. static final class Node<K,V> {  
2.         final K key;  
3.         volatile Object value;//value值  
4.         volatile Node<K,V> next;//next引用  
5.         ……  
6. }  
7. static class Index<K,V> {  
8.         final Node<K,V> node;  
9.         final Index<K,V> down;//downy引用  
10.        volatile Index<K,V> right;//右邊引用  
11.        ……  
12. }  

------------------------

ConcurrentSkipListMap的查找

經過SkipList的方式進行查找操做：（下圖以「查找91」進行說明：）

 

紅色虛線，表示查找的路徑，藍色向右箭頭表示right引用；黑色向下箭頭表示down引用；

 

/get方法，經過doGet操做實現

 

[java]  view plain copy

1. public V get(Object key) {  
2.       return doGet(key);  
3.  }  
4.  //doGet的實現  
5. private V doGet(Object okey) {  
6.         Comparable<? super K> key = comparable(okey);  
7.         Node<K,V> bound = null;  
8.         Index<K,V> q = head;//把頭結點做爲當前節點的前驅節點  
9.         Index<K,V> r = q.right;//前驅節點的右節點做爲當前節點  
10.         Node<K,V> n;  
11.         K k;  
12.         int c;  
13.         for (;;) {//遍歷  
14.             Index<K,V> d;  
15.             // 依次遍歷right節點  
16.             if (r != null && (n = r.node) != bound && (k = n.key) != null) {  
17.                 if ((c = key.compareTo(k)) > 0) {//因爲key都是升序排列的，全部當前關鍵字大於所要查找的key時繼續向右遍歷  
18.                     q = r;  
19.                     r = r.right;  
20.                     continue;  
21.                 } else if (c == 0) {  
22.                     //若是找到了相等的key節點，則返回該Node的value若是value爲空多是其餘併發delete致使的，因而經過另外一種  
23.                     //遍歷findNode的方式再查找  
24.                     Object v = n.value;  
25.                     return (v != null)? (V)v : getUsingFindNode(key);  
26.                 } else  
27.                     bound = n;  
28.             }  
29.             //若是一個鏈表中right沒能找到key對應的value，則調整到其down的引用處繼續查找  
30.             if ((d = q.down) != null) {  
31.                 q = d;  
32.                 r = d.right;  
33.             } else  
34.                 break;  
35.         }  
36.         // 若是經過上面的遍歷方式，還沒能找到key對應的value，再經過Node.next的方式進行查找  
37.         for (n = q.node.next;  n != null; n = n.next) {  
38.             if ((k = n.key) != null) {  
39.                 if ((c = key.compareTo(k)) == 0) {  
40.                     Object v = n.value;  
41.                     return (v != null)? (V)v : getUsingFindNode(key);  
42.                 } else if (c < 0)  
43.                     break;  
44.             }  
45.         }  
46.         return null;  
47.     }  

------------------------------------------------

ConcurrentSkipListMap的刪除

經過SkipList的方式進行刪除操做：（下圖以「刪除23」進行說明：）

 

紅色虛線，表示查找的路徑，藍色向右箭頭表示right引用；黑色向下箭頭表示down引用；

 

[java]  view plain copy

1. //remove操做，經過doRemove實現，把全部level中出現關鍵字key的地方都delete掉  
2. public V remove(Object key) {  
3.         return doRemove(key, null);  
4.  }  
5.  final V doRemove(Object okey, Object value) {  
6.         Comparable<? super K> key = comparable(okey);  
7.         for (;;) {  
8.             Node<K,V> b = findPredecessor(key);//獲得key的前驅（就是比key小的最大節點）  
9.             Node<K,V> n = b.next;//前驅節點的next引用  
10.             for (;;) {//遍歷  
11.                 if (n == null)//若是next引用爲空，直接返回  
12.                     return null;  
13.                 Node<K,V> f = n.next;  
14.                 if (n != b.next)                    // 若是兩次得到的b.next不是相同的Node，就跳轉到第一層循環從新得到b和n  
15.                     break;  
16.                 Object v = n.value;  
17.                 if (v == null) {                    // 當n被其餘線程delete的時候，其value==null，此時作輔助處理，並從新獲取b和n  
18.                     n.helpDelete(b, f);  
19.                     break;  
20.                 }  
21.                 if (v == n || b.value == null)      // 當其前驅被delet的時候直接跳出，從新獲取b和n  
22.                     break;  
23.                 int c = key.compareTo(n.key);  
24.                 if (c < 0)  
25.                     return null;  
26.                 if (c > 0) {//當key較大時就繼續遍歷  
27.                     b = n;  
28.                     n = f;  
29.                     continue;  
30.                 }  
31.                 if (value != null && !value.equals(v))  
32.                     return null;  
33.                 if (!n.casValue(v, null))  
34.                     break;  
35.                 if (!n.appendMarker(f) || !b.casNext(n, f))//casNext方法就是經過比較和設置b（前驅）的next節點的方式來實現刪除操做  
36.                     findNode(key);                  // 經過嘗試findNode的方式繼續find  
37.                 else {  
38.                     findPredecessor(key);           // Clean index  
39.                     if (head.right == null)   //若是head的right引用爲空，則表示不存在該level  
40.                         tryReduceLevel();  
41.                 }  
42.                 return (V)v;  
43.             }  
44.         }  
45.     }  

-------------------------------------

ConcurrentSkipListMap的插入

 

經過SkipList的方式進行插入操做：（下圖以「添加55」的兩種狀況，進行說明：）

在level=2（該level存在）的狀況下添加55的圖示:只需在level<=2的合適位置插入55便可

--------

在level=4(該level不存在，圖示level4是新建的)的狀況下添加55的狀況：首先新建level4,而後在level<=4的合適位置插入55

-----------

[java]  view plain copy

1. //put操做，經過doPut實現  
2.  public V put(K key, V value) {  
3.         if (value == null)  
4.             throw new NullPointerException();  
5.         return doPut(key, value, false);  
6.  }  
7. private V doPut(K kkey, V value, boolean onlyIfAbsent) {  
8.         Comparable<? super K> key = comparable(kkey);  
9.         for (;;) {  
10.             Node<K,V> b = findPredecessor(key);//前驅  
11.             Node<K,V> n = b.next;  
12.            //定位的過程就是和get操做類似  
13.             for (;;) {  
14.                 if (n != null) {  
15.                     Node<K,V> f = n.next;  
16.                     if (n != b.next)               // 先後值不一致的狀況下，跳轉到第一層循環從新得到b和n  
17.                         break;;  
18.                     Object v = n.value;  
19.                     if (v == null) {               // n被delete的狀況下  
20.                         n.helpDelete(b, f);  
21.                         break;  
22.                     }  
23.                     if (v == n || b.value == null) // b 被delete的狀況，從新獲取b和n  
24.                         break;  
25.                     int c = key.compareTo(n.key);  
26.                     if (c > 0) {  
27.                         b = n;  
28.                         n = f;  
29.                         continue;  
30.                     }  
31.                     if (c == 0) {  
32.                         if (onlyIfAbsent || n.casValue(v, value))  
33.                             return (V)v;  
34.                         else  
35.                             break; // restart if lost race to replace value  
36.                     }  
37.                     // else c < 0; fall through  
38.                 }  
39.                 Node<K,V> z = new Node<K,V>(kkey, value, n);  
40.                 if (!b.casNext(n, z))  
41.                     break;         // restart if lost race to append to b  
42.                 int level = randomLevel();//獲得一個隨機的level做爲該key-value插入的最高level  
43.                 if (level > 0)  
44.                     insertIndex(z, level);//進行插入操做  
45.                 return null;  
46.             }  
47.         }  
48.     }  
49.   
50.  /** 
51.      * 得到一個隨機的level值 
52.      */  
53.     private int randomLevel() {  
54.         int x = randomSeed;  
55.         x ^= x << 13;  
56.         x ^= x >>> 17;  
57.         randomSeed = x ^= x << 5;  
58.         if ((x & 0x8001) != 0) // test highest and lowest bits  
59.             return 0;  
60.         int level = 1;  
61.         while (((x >>>= 1) & 1) != 0) ++level;  
62.         return level;  
63.     }  
64. //執行插入操做：如上圖所示，有兩種可能的狀況：  
65. //1.當level存在時，對level<=n都執行insert操做  
66. //2.當level不存在（大於目前的最大level）時，首先添加新的level，而後在執行操做1   
67. private void insertIndex(Node<K,V> z, int level) {  
68.         HeadIndex<K,V> h = head;  
69.         int max = h.level;  
70.         if (level <= max) {//狀況1  
71.             Index<K,V> idx = null;  
72.             for (int i = 1; i <= level; ++i)//首先獲得一個包含1~level個級別的down關係的鏈表，最後的inx爲最高level  
73.                 idx = new Index<K,V>(z, idx, null);  
74.             addIndex(idx, h, level);//把最高level的idx傳給addIndex方法  
75.         } else { // 狀況2 增長一個新的級別  
76.             level = max + 1;  
77.             Index<K,V>[] idxs = (Index<K,V>[])new Index[level+1];  
78.             Index<K,V> idx = null;  
79.             for (int i = 1; i <= level; ++i)//該步驟和狀況1相似  
80.                 idxs[i] = idx = new Index<K,V>(z, idx, null);  
81.             HeadIndex<K,V> oldh;  
82.             int k;  
83.             for (;;) {  
84.                 oldh = head;  
85.                 int oldLevel = oldh.level;  
86.                 if (level <= oldLevel) { // lost race to add level  
87.                     k = level;  
88.                     break;  
89.                 }  
90.                 HeadIndex<K,V> newh = oldh;  
91.                 Node<K,V> oldbase = oldh.node;  
92.                 for (int j = oldLevel+1; j <= level; ++j)  
93.                     newh = new HeadIndex<K,V>(oldbase, newh, idxs[j], j);//建立新的  
94.                 if (casHead(oldh, newh)) {  
95.                     k = oldLevel;  
96.                     break;  
97.                 }  
98.             }  
99.             addIndex(idxs[k], oldh, k);  
100.         }  
101.     }  
102. /** 
103.      *在1~indexlevel層中插入數據  
104.      */  
105.     private void addIndex(Index<K,V> idx, HeadIndex<K,V> h, int indexLevel) {  
106.         //  insertionLevel 表明要插入的level，該值會在indexLevel~1間遍歷一遍  
107.         int insertionLevel = indexLevel;  
108.         Comparable<? super K> key = comparable(idx.node.key);  
109.         if (key == null) throw new NullPointerException();  
110.         // 和get操做相似，不一樣的就是查找的同時在各個level上加入了對應的key  
111.         for (;;) {  
112.             int j = h.level;  
113.             Index<K,V> q = h;  
114.             Index<K,V> r = q.right;  
115.             Index<K,V> t = idx;  
116.             for (;;) {  
117.                 if (r != null) {  
118.                     Node<K,V> n = r.node;  
119.                     // compare before deletion check avoids needing recheck  
120.                     int c = key.compareTo(n.key);  
121.                     if (n.value == null) {  
122.                         if (!q.unlink(r))  
123.                             break;  
124.                         r = q.right;  
125.                         continue;  
126.                     }  
127.                     if (c > 0) {  
128.                         q = r;  
129.                         r = r.right;  
130.                         continue;  
131.                     }  
132.                 }  
133.                 if (j == insertionLevel) {//在該層level中執行插入操做  
134.                     // Don't insert index if node already deleted  
135.                     if (t.indexesDeletedNode()) {  
136.                         findNode(key); // cleans up  
137.                         return;  
138.                     }  
139.                     if (!q.link(r, t))//執行link操做，其實就是inset的實現部分  
140.                         break; // restart  
141.                     if (--insertionLevel == 0) {  
142.                         // need final deletion check before return  
143.                         if (t.indexesDeletedNode())  
144.                             findNode(key);  
145.                         return;  
146.                     }  
147.                 }  
148.                 if (--j >= insertionLevel && j < indexLevel)//key移動到下一層level  
149.                     t = t.down;  
150.                 q = q.down;  
151.                 r = q.right;  
152.             }  
153.         }  
154.     }  

參考：

集合框架 Map篇（5）----ConcurrentSkipListMap http://hi.baidu.com/yao1111yao/item/0f3008163c4b82c938cb306d Java裏多個Map的性能比較（TreeMap、HashMap、ConcurrentSkipListMap） http://blog.hongtium.com/java-map-skiplist/ 跳錶SkipList的原理和實現 http://imtinx.iteye.com/blog/1291165