詳解Java同步容器和併發容器
同步容器
在 Java 中，同步容器主要包括 2 類：java
- Vector、Stack、HashTableCollections 類中提供的靜態工廠方法建立的類（由 Collections.synchronizedXxxx 等方法）
- Vector 實現了 List 接口，Vector 實際上就是一個數組，和 ArrayList 相似，可是 Vector 中的方法都是 synchronized 方法，即進行了同步措施。
- Stack 也是一個同步容器，它的方法也用 synchronized 進行了同步，它其實是繼承於 Vector 類。
- HashTable 實現了 Map 接口，它和 HashMap 很類似，可是 HashTable 進行了同步處理，而 HashMap 沒有。
同步容器的缺陷
同步容器的同步原理就是在方法上用 synchronized 修飾。那麼，這些方法每次只容許一個線程調用執行。算法
性能問題
因爲被 synchronized 修飾的方法，每次只容許一個線程執行，其餘試圖訪問這個方法的線程只能等待。顯然，這種方式比沒有使用 synchronized 的容器性能要差。數組
安全問題
同步容器真的必定安全嗎？安全
答案是：未必。同步容器未必真的安全。在作複合操做時，仍然須要加鎖來保護。bash
常見覆合操做以下：數據結構
- 迭代：反覆訪問元素，直到遍歷徹底部元素；
- 跳轉：根據指定順序尋找當前元素的下一個（下 n 個）元素；
- 條件運算：例如若沒有則添加等；
不安全的示例
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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) {
}
}
}
}
</pre>複製代碼
執行時可能會出現數組越界錯誤。併發
Vector 是線程安全的，爲何還會報這個錯？很簡單，對於 Vector，雖然能保證每個時刻只能有一個線程訪問它，可是不排除這種可能：性能
當某個線程在某個時刻執行這句時：學習
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">for (int i=0;i<vector.size();i++)
vector.get(i);
</pre>複製代碼
倘若此時 vector 的 size 方法返回的是 10，i 的值爲 9ui
而後另一個線程執行了這句：
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">for (int i=0;i<vector.size();i++)
vector.remove(i);
</pre>複製代碼
將下標爲 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，依次類推。
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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;
}
}
</pre>複製代碼
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)，能夠改進性能。
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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;
}
</pre>複製代碼
示例
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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);
}
}</pre>複製代碼
CopyOnWriteArrayList
要點
- 做用：CopyOnWrite 字面意思爲寫入時複製。CopyOnWriteArrayList 是線程安全的 ArrayList。
- 原理：
- 在 CopyOnWriteAarrayList 中，讀操做不一樣步，由於它們在內部數組的快照上工做，因此多個迭代器能夠同時遍歷而不會相互阻塞（1,2,4）。
- 全部的寫操做都是同步的。他們在備份數組（3）的副本上工做。寫操做完成後，後備陣列將被替換爲複製的陣列，並釋放鎖定。支持數組變得易變，因此替換數組的調用是原子（5）。
- 寫操做後建立的迭代器將可以看到修改的結構（6,7）。
- 寫時複製集合返回的迭代器不會拋出 ConcurrentModificationException，由於它們在數組的快照上工做，而且不管後續的修改（2,4）如何，都會像迭代器建立時那樣徹底返回元素。
源碼
重要屬性
- lock - 執行寫時複製操做，須要使用可重入鎖加鎖
- array - 對象數組，用於存放元素
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">
/** The lock protecting all mutators */
final transient ReentrantLock lock = new ReentrantLock();
/** The array, accessed only via getArray/setArray. */
private transient volatile Object[] array;
</pre>複製代碼
重要方法
- 添加操做
- 添加的邏輯很簡單，先將原容器 copy 一份，而後在新副本上執行寫操做，以後再切換引用。固然此過程是要加鎖的。
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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();
}
}
</pre>複製代碼
刪除操做
- 刪除操做同理，將除要刪除元素以外的其餘元素拷貝到新副本中，而後切換引用，將原容器引用指向新副本。同屬寫操做，須要加鎖。
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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();
}
}
</pre>複製代碼
- 讀操做
- CopyOnWriteArrayList 的讀操做是不用加鎖的，性能很高。
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">public E get(int index) {
return get(getArray(), index);
}
private E get(Object[] a, int index) {
return (E) a[index];
}
</pre>複製代碼
示例
<pre style="margin: 0px; padding: 0px; white-space: pre-wrap; overflow-wrap: break-word; font-family: "Courier New" !important; font-size: 12px !important;">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();
}
}
</pre>複製代碼
歡迎你們掃下方二維碼加java學習技術交流羣，一塊兒夯實基礎，提高自我價值
相關文章
- 1. JAVA同步容器和併發容器
- 2. Java同步容器和併發容器
- 3. Java併發編程(4)-同步容器和併發容器
- 4. 同步容器與併發容器(轉)
- 5. Java併發（9）- 從同步容器到併發容器
- 6. 【Java併發編程】同步容器與併發容器
- 7. Java併發-從同步容器到併發容器
- 8. java併發：同步容器&併發容器
- 9. java 普通容器，同步容器，併發容器，同步工具
- 10. Java多線程併發之同步容器和併發容器-第一篇
- 更多相關文章...
- • Spring IoC容器：BeanFactory和ApplicationContext - Spring教程
- • Docker 容器連接 - Docker教程
- • Docker容器實戰(七) - 容器眼光下的文件系統
- • Docker容器實戰(六) - 容器的隔離與限制
相關標籤/搜索
每日一句
-
每一个你不满意的现在，都有一个你没有努力的曾经。
歡迎關注本站公眾號,獲取更多信息
