Java同步容器和併發容器
同步容器
在 Java 中,同步容器主要包括 2 類:
- Vector、Stack、HashTable
- Vector 實現了 List 介面,Vector 實際上就是一個數組,和 ArrayList 類似,但是 Vector 中的方法都是 synchronized 方法,即進行了同步措施。
- Stack 也是一個同步容器,它的方法也用 synchronized 進行了同步,它實際上是繼承於 Vector 類。
- HashTable 實現了 Map 介面,它和 HashMap 很相似,但是 HashTable 進行了同步處理,而 HashMap 沒有。
- Collections 類中提供的靜態工廠方法建立的類(由 Collections.synchronizedXxxx 等方法)
同步容器的缺陷
同步容器的同步原理就是在方法上用 synchronized
修飾。那麼,這些方法每次只允許一個執行緒呼叫執行。
效能問題
由於被 synchronized
修飾的方法,每次只允許一個執行緒執行,其他試圖訪問這個方法的執行緒只能等待。顯然,這種方式比沒有使用 synchronized
安全問題
同步容器真的一定安全嗎?
答案是:未必。同步容器未必真的安全。在做複合操作時,仍然需要加鎖來保護。
常見覆合操作如下:
- 迭代:反覆訪問元素,直到遍歷完全部元素;
- 跳轉:根據指定順序尋找當前元素的下一個(下 n 個)元素;
- 條件運算:例如若沒有則新增等;
不安全的示例
public class Test { static Vector<Integer> vector = new Vector<Integer>(); public static void main(String[] args) throws InterruptedException { while(true) { for(int i=0;i<10;i++) vector.add(i); Thread thread1 = new Thread(){ public void run() { for(int i=0;i<vector.size();i++) vector.remove(i); }; }; Thread thread2 = new Thread(){ public void run() { for(int i=0;i<vector.size();i++) vector.get(i); }; }; thread1.start(); thread2.start(); while(Thread.activeCount()>10) { } } } }
執行時可能會出現陣列越界錯誤。
Vector 是執行緒安全的,為什麼還會報這個錯?很簡單,對於 Vector,雖然能保證每一個時刻只能有一個執行緒訪問它,但是不排除這種可能:
當某個執行緒在某個時刻執行這句時:
for(int i=0;i<vector.size();i++) vector.get(i);
假若此時 vector 的 size 方法返回的是 10,i 的值為 9
然後另外一個執行緒執行了這句:
for(int i=0;i<vector.size();i++) vector.remove(i);
將下標為 9 的元素刪除了。
那麼通過 get 方法訪問下標為 9 的元素肯定就會出問題了。
安全示例
因此為了保證執行緒安全,必須在方法呼叫端做額外的同步措施,如下面所示:
public class Test { static Vector<Integer> vector = new Vector<Integer>(); public static void main(String[] args) throws InterruptedException { while(true) { for(int i=0;i<10;i++) vector.add(i); Thread thread1 = new Thread(){ public void run() { synchronized (Test.class) { //進行額外的同步 for(int i=0;i<vector.size();i++) vector.remove(i); } }; }; Thread thread2 = new Thread(){ public void run() { synchronized (Test.class) { for(int i=0;i<vector.size();i++) vector.get(i); } }; }; thread1.start(); thread2.start(); while(Thread.activeCount()>10) { } } } }
ConcurrentModificationException 異常
在對 Vector 等容器併發地進行迭代修改時,會報 ConcurrentModificationException 異常,關於這個異常將會在後續文章中講述。
但是在併發容器中不會出現這個問題。
併發容器
JDK 的 java.util.concurrent
包(即 juc)中提供了幾個非常有用的併發容器。
- CopyOnWriteArrayList - 執行緒安全的 ArrayList
- CopyOnWriteArraySet - 執行緒安全的 Set,它內部包含了一個 CopyOnWriteArrayList,因此本質上是由 CopyOnWriteArrayList 實現的。
- ConcurrentSkipListSet - 相當於執行緒安全的 TreeSet。它是有序的 Set。它由 ConcurrentSkipListMap 實現。
- ConcurrentHashMap - 執行緒安全的 HashMap。採用分段鎖實現高效併發。
- ConcurrentSkipListMap - 執行緒安全的有序 Map。使用跳錶實現高效併發。
- ConcurrentLinkedQueue - 執行緒安全的無界佇列。底層採用單鏈表。支援 FIFO。
- ConcurrentLinkedDeque - 執行緒安全的無界雙端佇列。底層採用雙向連結串列。支援 FIFO 和 FILO。
- ArrayBlockingQueue - 陣列實現的阻塞佇列。
- LinkedBlockingQueue - 連結串列實現的阻塞佇列。
- LinkedBlockingDeque - 雙向連結串列實現的雙端阻塞佇列。
ConcurrentHashMap
要點
- 作用:ConcurrentHashMap 是執行緒安全的 HashMap。
- 原理:JDK6 與 JDK7 中,ConcurrentHashMap 採用了分段鎖機制。JDK8 中,摒棄了鎖分段機制,改為利用 CAS 演算法。
原始碼
JDK7
ConcurrentHashMap 類在 jdk1.7 中的設計,其基本結構如圖所示:
每一個 segment 都是一個 HashEntry<K,V>[] table, table 中的每一個元素本質上都是一個 HashEntry 的單向佇列。比如 table[3]為首節點,table[3]->next 為節點 1,之後為節點 2,依次類推。
public class ConcurrentHashMap<K, V> extends AbstractMap<K, V> implements ConcurrentMap<K, V>, Serializable { // 將整個hashmap分成幾個小的map,每個segment都是一個鎖;與hashtable相比,這麼設計的目的是對於put, remove等操作,可以減少併發衝突,對 // 不屬於同一個片段的節點可以併發操作,大大提高了效能 final Segment<K,V>[] segments; // 本質上Segment類就是一個小的hashmap,裡面table陣列儲存了各個節點的資料,繼承了ReentrantLock, 可以作為互拆鎖使用 static final class Segment<K,V> extends ReentrantLock implements Serializable { transient volatile HashEntry<K,V>[] table; transient int count; } // 基本節點,儲存Key, Value值 static final class HashEntry<K,V> { final int hash; final K key; volatile V value; volatile HashEntry<K,V> next; } }
JDK8
- jdk8 中主要做了 2 方面的改進
- 取消 segments 欄位,直接採用
transient volatile HashEntry<K,V>[] table
儲存資料,採用 table 陣列元素作為鎖,從而實現了對每一行資料進行加鎖,進一步減少併發衝突的概率。 - 將原先 table 陣列+單向連結串列的資料結構,變更為 table 陣列+單向連結串列+紅黑樹的結構。對於 hash 表來說,最核心的能力在於將 key hash 之後能均勻的分佈在陣列中。如果 hash 之後雜湊的很均勻,那麼 table 陣列中的每個佇列長度主要為 0 或者 1。但實際情況並非總是如此理想,雖然 ConcurrentHashMap 類預設的載入因子為 0.75,但是在資料量過大或者運氣不佳的情況下,還是會存在一些佇列長度過長的情況,如果還是採用單向列表方式,那麼查詢某個節點的時間複雜度為 O(n);因此,對於個數超過 8(預設值)的列表,jdk1.8 中採用了紅黑樹的結構,那麼查詢的時間複雜度可以降低到 O(logN),可以改進效能。
final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); int hash = spread(key.hashCode()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; // 如果table為空,初始化;否則,根據hash值計算得到陣列索引i,如果tab[i]為空,直接新建節點Node即可。注:tab[i]實質為連結串列或者紅黑樹的首節點。 if (tab == null || (n = tab.length) == 0) tab = initTable(); else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } // 如果tab[i]不為空並且hash值為MOVED,說明該連結串列正在進行transfer操作,返回擴容完成後的table。 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; // 針對首個節點進行加鎖操作,而不是segment,進一步減少執行緒衝突 synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; // 如果在連結串列中找到值為key的節點e,直接設定e.val = value即可。 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } // 如果沒有找到值為key的節點,直接新建Node並加入連結串列即可。 Node<K,V> pred = e; if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } // 如果首節點為TreeBin型別,說明為紅黑樹結構,執行putTreeVal操作。 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) { // 如果節點數>=8,那麼轉換連結串列結構為紅黑樹結構。 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } // 計數增加1,有可能觸發transfer操作(擴容)。 addCount(1L, binCount); return null; }
示例
public class ConcurrentHashMapDemo { public static void main(String[] args) throws InterruptedException { // HashMap 在併發迭代訪問時會丟擲 ConcurrentModificationException 異常 // Map<Integer, Character> map = new HashMap<>(); Map<Integer, Character> map = new ConcurrentHashMap<>(); Thread wthread = new Thread(() -> { System.out.println("寫操作執行緒開始執行"); for (int i = 0; i < 26; i++) { map.put(i, (char) ('a' + i)); } }); Thread rthread = new Thread(() -> { System.out.println("讀操作執行緒開始執行"); for (Integer key : map.keySet()) { System.out.println(key + " - " + map.get(key)); } }); wthread.start(); rthread.start(); Thread.sleep(1000); } }
CopyOnWriteArrayList
要點
- 作用:CopyOnWrite 字面意思為寫入時複製。CopyOnWriteArrayList 是執行緒安全的 ArrayList。
- 原理:
- 在 CopyOnWriteAarrayList 中,讀操作不同步,因為它們在內部陣列的快照上工作,所以多個迭代器可以同時遍歷而不會相互阻塞(1,2,4)。
- 所有的寫操作都是同步的。他們在備份陣列(3)的副本上工作。寫操作完成後,後備陣列將被替換為複製的陣列,並釋放鎖定。支援陣列變得易變,所以替換陣列的呼叫是原子(5)。
- 寫操作後建立的迭代器將能夠看到修改的結構(6,7)。
- 寫時複製集合返回的迭代器不會丟擲 ConcurrentModificationException,因為它們在陣列的快照上工作,並且無論後續的修改(2,4)如何,都會像迭代器建立時那樣完全返回元素。
原始碼
重要屬性
- lock - 執行寫時複製操作,需要使用可重入鎖加鎖
- array - 物件陣列,用於存放元素
/** The lock protecting all mutators */ final transient ReentrantLock lock = new ReentrantLock(); /** The array, accessed only via getArray/setArray. */ private transient volatile Object[] array;
重要方法
- 新增操作
- 新增的邏輯很簡單,先將原容器 copy 一份,然後在新副本上執行寫操作,之後再切換引用。當然此過程是要加鎖的。
public boolean add(E e) { //ReentrantLock加鎖,保證執行緒安全 final ReentrantLock lock = this.lock; lock.lock(); try { Object[] elements = getArray(); int len = elements.length; //拷貝原容器,長度為原容器長度加一 Object[] newElements = Arrays.copyOf(elements, len + 1); //在新副本上執行新增操作 newElements[len] = e; //將原容器引用指向新副本 setArray(newElements); return true; } finally { //解鎖 lock.unlock(); } }
- 刪除操作
- 刪除操作同理,將除要刪除元素之外的其他元素拷貝到新副本中,然後切換引用,將原容器引用指向新副本。同屬寫操作,需要加鎖。
public E remove(int index) { //加鎖 final ReentrantLock lock = this.lock; lock.lock(); try { Object[] elements = getArray(); int len = elements.length; E oldValue = get(elements, index); int numMoved = len - index - 1; if (numMoved == 0) //如果要刪除的是列表末端資料,拷貝前len-1個數據到新副本上,再切換引用 setArray(Arrays.copyOf(elements, len - 1)); else { //否則,將除要刪除元素之外的其他元素拷貝到新副本中,並切換引用 Object[] newElements = new Object[len - 1]; System.arraycopy(elements, 0, newElements, 0, index); System.arraycopy(elements, index + 1, newElements, index, numMoved); setArray(newElements); } return oldValue; } finally { //解鎖 lock.unlock(); } }
- 讀操作
- CopyOnWriteArrayList 的讀操作是不用加鎖的,效能很高。
public E get(int index) { return get(getArray(), index); } private E get(Object[] a, int index) { return (E) a[index]; }
示例
public class CopyOnWriteArrayListDemo { static class ReadTask implements Runnable { List<String> list; ReadTask(List<String> list) { this.list = list; } public void run() { for (String str : list) { System.out.println(str); } } } static class WriteTask implements Runnable { List<String> list; int index; WriteTask(List<String> list, int index) { this.list = list; this.index = index; } public void run() { list.remove(index); list.add(index, "write_" + index); } } public void run() { final int NUM = 10; // ArrayList 在併發迭代訪問時會丟擲 ConcurrentModificationException 異常 // List<String> list = new ArrayList<>(); CopyOnWriteArrayList<String> list = new CopyOnWriteArrayList<>(); for (int i = 0; i < NUM; i++) { list.add("main_" + i); } ExecutorService executorService = Executors.newFixedThreadPool(NUM); for (int i = 0; i < NUM; i++) { executorService.execute(new ReadTask(list)); executorService.execute(new WriteTask(list, i)); } executorService.shutdown(); } public static void main(String[] args) { new CopyOnWriteArrayListDemo().run(); } }
免費Java資料需要自己領取,涵蓋了Java、Redis、MongoDB、MySQL、Zookeeper、Spring Cloud、Dubbo高併發分散式等教程。
傳送門:https://mp.weixin.qq.com/s/JzddfH-7yNudmkjT0IRL