Java并發容器，底層原理深入分析

ConcurrentHashMap

ConcurrentHashMap底層具體實作

JDK 1.7底層實作

将資料分為一段一段的存儲，然後給每一段資料配一把鎖， 當一個線程占用鎖通路其中一個段資料時，其他段的資料也能被其他線程通路。

ConcurrentHashMap是由Segment數組結構和HashEntry數組結構組成。 其中Segment 實作了 ReentrantLock,是以Segment是一種可重入鎖，扮演鎖的角色。 HashEntry 用于存儲鍵值對資料。

一個ConcurrentHashMap裡包含一個Segment數組。 Segment結構和HashMap類似，是一種數組和連結清單結構， 一個Segment包含一個HashEntry數組，每個HashEntry是一個連結清單結構的元素， 每個Segment守護着一個HashEntry數組裡的元素，當對HashEntry數組的資料進行修改時，必須首先獲得對應的Segment的鎖。

JDK 1.8底層實作

TreeBin: 紅黑二叉樹節點 Node: 連結清單節點

v2-9e5e57b2227df02dc231978862ff5cf8_b.png

ConcurrentHashMap取消了Segment分段鎖，采用CAS和synchronized來保證并發安全。 資料結構與HashMap1.8的結構類似，數組+連結清單/紅黑二叉樹(連結清單長度>8時，轉換為紅黑樹)。

synchronized隻鎖定目前連結清單或紅黑二叉樹的首節點，這樣隻要hash值不沖突，就不會産生并發。

ConcurrentHashMap和Hashtable的差別

底層資料結構：

JDK1.7 的ConcurrentHashMap底層采用分段的數組+連結清單實作， JDK1.8 的ConcurrentHashMap底層采用的資料結構與JDK1.8 的HashMap的結構一樣，數組+連結清單/紅黑二叉樹。

Hashtable和JDK1.8 之前的HashMap的底層資料結構類似都是采用數組+連結清單的形式， 數組是 HashMap 的主體，連結清單則是主要為了解決哈希沖突而存在的

實作線程安全的方式

JDK1.7的ConcurrentHashMap（分段鎖）對整個桶數組進行了分割分段(Segment)， 每一把鎖隻鎖容器其中一部分資料，多線程通路容器裡不同資料段的資料，就不會存在鎖競争，提高并發通路率。 JDK 1.8 采用數組+連結清單/紅黑二叉樹的資料結構來實作，并發控制使用synchronized和CAS來操作。

Hashtable:使用 synchronized 來保證線程安全，效率非常低下。 當一個線程通路同步方法時，其他線程也通路同步方法，可能會進入阻塞或輪詢狀态， 如使用 put 添加元素，另一個線程不能使用 put 添加元素，也不能使用 get，競争會越來越激烈。

HashTable全表鎖

v2-72e2203eeb5a463c4a0fd6295ce6fc8a_b.png

ConcurrentHashMap分段鎖

v2-658c1f06db5fe9820c8c15e151462e7a_b.png

CopyOnWriteArrayList

public class CopyOnWriteArrayList extends Object

implements List, RandomAccess, Cloneable, Serializable

在很多應用場景中，讀操作可能會遠遠大于寫操作。 由于讀操作根本不會修改原有的資料，是以對于每次讀取都進行加鎖其實是一種資源浪費。 我們應該允許多個線程同時通路List的内部資料，畢竟讀取操作是安全的。 這和ReentrantReadWriteLock讀寫鎖的思想非常類似，也就是讀讀共享、寫寫互斥、讀寫互斥、寫讀互斥。 JDK中提供了CopyOnWriteArrayList類比相比于在讀寫鎖的思想又更進一步。 為了将讀取的性能發揮到極緻，CopyOnWriteArrayList 讀取是完全不用加鎖的，并且寫入也不會阻塞讀取操作。 隻有寫入和寫入之間需要進行同步等待。這樣，讀操作的性能就會大幅度提高。

CopyOnWriteArrayList的實作機制

CopyOnWriteArrayLis 類的所有可變操作（add，set等等）都是通過建立底層數組的新副本來實作的。 當 List 需要被修改的時候，我并不修改原有内容，而是對原有資料進行一次複制，将修改的内容寫入副本。 寫完之後，再将修改完的副本替換原來的資料，這樣就可以保證寫操作不會影響讀操作了。

CopyOnWriteArrayList是滿足CopyOnWrite的ArrayList， 所謂CopyOnWrite也就是說： 在計算機，如果你想要對一塊記憶體進行修改時，我們不在原有記憶體塊中進行寫操作， 而是将記憶體拷貝一份，在新的記憶體中進行寫操作，寫完之後呢，就将指向原來記憶體指針指向新的記憶體， 原來的記憶體就可以被回收掉了。

CopyOnWriteArrayList讀取和寫入源碼簡單分析

CopyOnWriteArrayList讀取操作的實作

讀取操作沒有任何同步控制和鎖操作， 因為内部數組array不會發生修改，隻會被另外一個array替換，是以可以保證資料安全。

/* The array, accessed only via getArray/setArray. /

private transient volatile Object[] array;

public E get(int index) {

return get(getArray(), index);           

}

@SuppressWarnings("unchecked")

private E get(Object[] a, int index) {

return (E) a[index];           

final Object[] getArray() {

return array;           

CopyOnWriteArrayList 寫入操作 add() 方法在添加集合的時候加了鎖， 保證同步，避免了多線程寫的時候會複制出多個副本出來。

/**

• Appends the specified element to the end of this list.

  */

public boolean add(E e) {

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();//釋放鎖
}           

ConcurrentLinkedQueue

Java提供的線程安全的 Queue 可以分為阻塞隊列和非阻塞隊列，其中阻塞隊列的典型例子是 BlockingQueue， 非阻塞隊列的典型例子是ConcurrentLinkedQueue，在實際應用中要根據實際需要選用阻塞隊列或者非阻塞隊列。 阻塞隊列可以通過加鎖來實作，非阻塞隊列可以通過CAS操作實作。

ConcurrentLinkedQueue使用連結清單作為其資料結構。 ConcurrentLinkedQueue應該算是在高并發環境中性能最好的隊列了。 它之所有能有很好的性能，是因為其内部複雜的實作。

ConcurrentLinkedQueue 主要使用CAS非阻塞算法來實作線程安全。 适合在對性能要求相對較高，對個線程同時對隊列進行讀寫的場景， 即如果對隊列加鎖的成本較高則适合使用無鎖的ConcurrentLinkedQueue來替代。

BlockingQueue

java.util.concurrent.BlockingQueue 接口有以下阻塞隊列的實作：

FIFO 隊列 ：LinkedBlockingQueue、ArrayBlockingQueue（固定長度）

優先級隊列 ：PriorityBlockingQueue

提供了阻塞的 take() 和 put() 方法：如果隊列為空 take() 将阻塞，直到隊列中有内容； 如果隊列為滿 put() 将阻塞，直到隊列有空閑位置。

使用 BlockingQueue 實作生産者消費者問題

public class ProducerConsumer {

private static BlockingQueue<String> queue = new ArrayBlockingQueue<>(5);

private static class Producer extends Thread {
    @Override
    public void run() {
        try {
            queue.put("product");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.print("produce..");
    }
}

private static class Consumer extends Thread {

    @Override
    public void run() {
        try {
            String product = queue.take();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.print("consume..");
    }
}           

public static void main(String[] args) {

for (int i = 0; i < 2; i++) {
    Producer producer = new Producer();
    producer.start();
}
for (int i = 0; i < 5; i++) {
    Consumer consumer = new Consumer();
    consumer.start();
}
for (int i = 0; i < 3; i++) {
    Producer producer = new Producer();
    producer.start();
}           

produce..produce..consume..consume..produce..consume..produce..consume..produce..consume