Java併發（9）- 從同步容器到併發容器
引言
容器是Java基礎類庫中使用頻率最高的一部分，Java集合包中提供了大量的容器類來幫組我們簡化開發，我前面的文章中對Java集合包中的關鍵容器進行過一個系列的分析，但這些集合類都是非執行緒安全的，即在多執行緒的環境下，都需要其他額外的手段來保證資料的正確性，最簡單的就是通過synchronized關鍵字將所有使用到非執行緒安全的容器程式碼全部同步執行。這種方式雖然可以達到執行緒安全的目的，但存在幾個明顯的問題：首先編碼上存在一定的複雜性，相關的程式碼段都需要新增鎖。其次這種一刀切的做法在高併發情況下效能並不理想，基本相當於序列執行。JDK1.5中為我們提供了一系列的併發容器，集中在java.util.concurrent包下，用來解決這兩個問題，先從同步容器說起。
同步容器Vector和HashTable
為了簡化程式碼開發的過程，早期的JDK在java.util包中提供了Vector和HashTable兩個同步容器，這兩個容器的實現和早期的ArrayList和HashMap程式碼實現基本一樣，不同在於Vector和HashTable在每個方法上都添加了synchronized關鍵字來保證同一個例項同時只有一個執行緒能訪問，部分原始碼如下：
//Vector public synchronized int size() {}; public synchronized E get(int index) {}; //HashTable public synchronized V put(K key, V value) {}; public synchronized V remove(Object key) {};
通過對每個方法新增synchronized，保證了多次操作的序列。這種方式雖然使用起來方便了，但並沒有解決高併發下的效能問題，與手動鎖住ArrayList和HashMap並沒有什麼區別，不論讀還是寫都會鎖住整個容器。其次這種方式存在另一個問題：當多個執行緒進行復合操作時，是執行緒不安全的。可以通過下面的程式碼來說明這個問題：
public static void deleteVector(){
int index = vectors.size() - 1;
vectors.remove(index);
}程式碼中對Vector進行了兩步操作，首先獲取size，然後移除最後一個元素，多執行緒情況下如果兩個執行緒交叉執行，A執行緒呼叫size後，B執行緒移除最後一個元素，這時A執行緒繼續remove將會丟擲索引超出的錯誤。
那麼怎麼解決這個問題呢？最直接的修改方案就是對程式碼塊加鎖來防止多執行緒同時執行：
public static void deleteVector(){
synchronized (vectors) {
int index = vectors.size() - 1;
vectors.remove(index);
}
}如果上面的問題通過加鎖來解決沒有太直觀的影響，那麼來看看對vectors進行迭代的情況：
public static void foreachVector(){
synchronized (vectors) {
for (int i = 0; i < vectors.size(); i++) {
System.out.println(vectors.get(i).toString());
}
}
}為了避免多執行緒情況下在迭代的過程中其他執行緒對vectors進行了修改，就不得不對整個迭代過程加鎖，想象這麼一個場景，如果迭代操作非常頻繁，或者vectors元素很大，那麼所有的修改和讀取操作將不得不在鎖外等待，這將會對多執行緒效能造成極大的影響。那麼有沒有什麼方式能夠很好的對容器的迭代操作和修改操作進行分離，在修改時不影響容器的迭代操作呢？這就需要java.util.concurrent包中的各種併發容器了出場了。
併發容器CopyOnWrite
CopyOnWrite--寫時複製容器是一種常用的併發容器，它通過多執行緒下讀寫分離來達到提高併發效能的目的，和前面我們講解StampedLock時所用的解決方案類似：任何時候都可以進行讀操作，寫操作則需要加鎖。不同的是，在CopyOnWrite中，對容器的修改操作加鎖後，通過copy一個新的容器來進行修改，修改完畢後將容器替換為新的容器即可。
這種方式的好處顯而易見：通過copy一個新的容器來進行修改，這樣讀操作就不需要加鎖，可以併發讀，因為在讀的過程中是採用的舊的容器，即使新容器做了修改對舊容器也沒有影響，同時也很好的解決了迭代過程中其他執行緒修改導致的併發問題。
JDK中提供的併發容器包括CopyOnWriteArrayList和CopyOnWriteArraySet，下面通過CopyOnWriteArrayList的部分原始碼來理解這種思想：
//新增元素
public boolean add(E e) {
//獨佔鎖
final ReentrantLock lock = this.lock;
lock.lock();
try {
Object[] elements = getArray();
int len = elements.length;
//複製一個新的陣列newElements
Object[] newElements = Arrays.copyOf(elements, len + 1);
newElements[len] = e;
//修改後指向新的陣列
setArray(newElements);
return true;
} finally {
lock.unlock();
}
}
public E get(int index) {
//未加鎖，直接獲取
return get(getArray(), index);
}程式碼很簡單，在add操作中通過一個共享的ReentrantLock來獲取鎖，這樣可以防止多執行緒下多個執行緒同時修改容器內容。獲取鎖後通過Arrays.copyOf複製了一個新的容器，然後對新的容器進行了修改，最後直接通過setArray將原陣列引用指向了新的陣列，避免了在修改過程中迭代資料出現錯誤。get操作由於是讀操作，未加鎖，直接讀取就行。CopyOnWriteArraySet類似，這裡不做過多講解。
CopyOnWrite容器雖然在多執行緒下使用是安全的，相比較Vector也大大提高了讀寫的效能，但它也有自身的問題。
首先就是效能，在講解ArrayList的文章中提到過，ArrayList的擴容由於使用了Arrays.copyOf每次都需要申請更大的空間以及複製現有的元素到新的陣列，對效能存在一定影響。CopyOnWrite容器也不例外，每次修改操作都會申請新的陣列空間，然後進行替換。所以在高併發頻繁修改容器的情況下，會不斷申請新的空間，同時會造成頻繁的GC，這時使用CopyOnWrite容器並不是一個好的選擇。
其次還有一個數據一致性問題，由於在修改中copy了新的陣列進行替換，同時舊陣列如果還在被使用，那麼新的資料就不能被及時讀取到，這樣就造成了資料不一致，如果需要強資料一致性，CopyOnWrite容器也不太適合。
併發容器ConcurrentHashMap
ConcurrentHashMap容器相較於CopyOnWrite容器在併發加鎖粒度上有了更大一步的優化，它通過修改對單個hash桶元素加鎖的達到了更細粒度的併發控制。在瞭解ConcurrentHashMap容器之前，推薦大家先閱讀我之前對HashMap原始碼分析的文章--Java集合（5）一 HashMap與HashSet，因為在底層資料結構上，ConcurrentHashMap和HashMap都使用了陣列+連結串列+紅黑樹的方式，只是在HashMap的基礎上添加了併發相關的一些控制，所以這裡只對ConcurrentHashMap中併發相關程式碼做一些分析。
還是先從ConcurrentHashMap的寫操作開始，這裡就是put方法：
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode()); //計算桶的hash值
int binCount = 0;
//迴圈插入元素，避免併發插入失敗
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
if (tab == null || (n = tab.length) == 0)
tab = initTable();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果當前桶無元素，則通過cas操作插入新節點
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break;
}
//如果當前桶正在擴容，則協助擴容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//hash衝突時鎖住當前需要新增節點的頭元素，可能是連結串列頭節點或者紅黑樹的根節點
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}在put元素的過程中，有幾個併發處理的關鍵點：
- 如果當前桶對應的節點還沒有元素插入，通過典型的無鎖cas操作嘗試插入新節點，減少加鎖的概率，併發情況下如果插入不成功，很容易想到自旋，也就是
for (Node<K,V>[] tab = table;;)。 - 如果當前桶正在擴容，則協助擴容
((fh = f.hash) == MOVED)。這裡是一個重點，ConcurrentHashMap的擴容和HashMap不一樣，它在多執行緒情況下或使用多個執行緒同時擴容，每個執行緒擴容指定的一部分hash桶，當前執行緒擴容完指定桶之後會繼續獲取下一個擴容任務，直到擴容全部完成。擴容的大小和HashMap一樣，都是翻倍，這樣可以有效減少移動的元素數量，也就是使用2的冪次方的原因，在HashMap中也一樣。 - 在發生hash衝突時僅僅只鎖住當前需要新增節點的頭元素即可，可能是連結串列頭節點或者紅黑樹的根節點，其他桶節點都不需要加鎖，大大減小了鎖粒度。
通過ConcurrentHashMap新增元素的過程，知道了ConcurrentHashMap容器是通過CAS + synchronized一起來實現併發控制的。這裡有個額外的問題：為什麼使用synchronized而不使用ReentrantLock？前面我的文章也對synchronized以及ReentrantLock的實現方式和效能做過分析，在這裡我的理解是synchronized在後期優化空間上比ReentrantLock更大。
併發容器ConcurrentSkipListMap
java.util中對應的容器在java.util.concurrent包中基本都可以找到對應的併發容器：List和Set有對應的CopyOnWriteArrayList與CopyOnWriteArraySet，HashMap有對應的ConcurrentHashMap，但是有序的TreeMap或並沒有對應的ConcurrentTreeMap。
為什麼沒有ConcurrentTreeMap呢？這是因為TreeMap內部使用了紅黑樹來實現，紅黑樹是一種自平衡的二叉樹，當樹被修改時，需要重新平衡，重新平衡操作可能會影響樹的大部分節點，如果併發量非常大的情況下，這就需要在許多樹節點上新增互斥鎖，那併發就失去了意義。所以提供了另外一種併發下的有序map實現：ConcurrentSkipListMap。
ConcurrentSkipListMap內部使用跳錶（SkipList）這種資料結構來實現，他的結構相對紅黑樹來說非常簡單理解，實現起來也相對簡單，而且在理論上它的查詢、插入、刪除時間複雜度都為log(n)。在併發上，ConcurrentSkipListMap採用無鎖的CAS+自旋來控制。
跳錶簡單來說就是一個多層的連結串列，底層是一個普通的連結串列，然後逐層減少，通常通過一個簡單的演算法實現每一層元素是下一層的元素的二分之一，這樣當搜尋元素時從最頂層開始搜尋，可以說是另一種形式的二分查詢。
一個簡單的獲取跳錶層數概率演算法實現如下：
int random_level() {
K = 1;
while (random(0,1))
K++;
return K;
} 通過簡單的0和1獲取概率，1層的概率為50%，2層的概率為25%，3層的概率為12.5%，這樣逐級遞減。
一個三層的跳錶新增元素的過程如下：
插入值為15的節點：
插入後：
維基百科中有一個新增節點的動圖，這裡也貼出來方便理解：
通過分析ConcurrentSkipListMap的put方法來理解跳錶以及CAS自旋併發控制：
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 (;;) {
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) //發生競爭，兩次節點獲取不一致，併發導致
break;
if ((v = n.value) == null) { // 節點已經被刪除
n.helpDelete(b, f);
break;
}
if (b.value == null || v == n)
break;
if ((c = cpr(cmp, key, n.key)) > 0) { //進行下一輪查詢，比當前key大
b = n;
n = f;
continue;
}
if (c == 0) { //相等時直接cas修改值
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
}
z = new Node<K,V>(key, value, n); //9. n.key > key > b.key
if (!b.casNext(n, z)) //cas修改值
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;
while (((rnd >>>= 1) & 1) != 0) // 獲取跳錶層級
++level;
Index<K,V> idx = null;
HeadIndex<K,V> h = head;
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，在這裡體現為Index<K,V>陣列
level = max + 1; // hold in array and later pick the one to use
@SuppressWarnings("unchecked")Index<K,V>[] idxs =
(Index<K,V>[])new Index<?,?>[level+1];
for (int i = 1; i <= level; ++i)
idxs[i] = idx = new Index<K,V>(z, idx, null);
for (;;) {
h = 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) //新新增的level層的具體資料
newh = new HeadIndex<K,V>(oldbase, newh, idxs[j], j);
if (casHead(h, newh)) {
h = newh;
idx = idxs[level = oldLevel];
break;
}
}
}
// 逐層插入資料過程
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;
}這裡的插入方法很複雜，可以分為3大步來理解：第一步獲取前繼節點後通過CAS來插入節點；第二步對level層數進行判斷，如果大於最大層數，則插入一層；第三步插入對應層的資料。整個插入過程全部通過CAS自旋的方式保證併發情況下的資料正確性。
總結
JDK中提供了豐富的併發容器供我們使用，文章中介紹的也並不全面，重點是要通過了解各種併發容器的原理，明白他們各自獨特的使用場景。這裡簡單做個總結：當併發讀遠多於修改的場景下需要使用List和Set時，可以考慮使用CopyOnWriteArrayList和CopyOnWriteArraySet；當需要併發使用<Key, Value>鍵值對存取資料時，可以使用ConcurrentHashMap；當要保證併發<Key, Value>鍵值對有序時可以使用ConcurrentSkipListMap。