Java同步容器和並發容器
阿新 • • 發佈：2019-04-05
-c moved 陣列 vol 鏈表的數據結構 mys zook 列表 segments
同步容器
在 Java 中，同步容器主要包括 2 類：
- Vector、Stack、HashTableCollections 類中提供的靜態工廠方法創建的類（由 Collections.synchronizedXxxx 等方法）
- Vector 實現了 List 接口，Vector 實際上就是一個數組，和 ArrayList 類似，但是 Vector 中的方法都是 synchronized 方法，即進行了同步措施。
- Stack 也是一個同步容器，它的方法也用 synchronized 進行了同步，它實際上是繼承於 Vector 類。
- HashTable 實現了 Map 接口，它和 HashMap 很相似，但是 HashTable 進行了同步處理，而 HashMap 沒有。
同步容器的缺陷
同步容器的同步原理就是在方法上用 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-7yNudmkjT0IRL8Q
Java同步容器和並發容器