作者：acupt，專注Java，架構(gòu)師社區(qū)合伙人！

14個(gè)并發(fā)容器，你用過(guò)幾個(gè)？

不考慮多線程并發(fā)的情況下，容器類一般使用ArrayList、HashMap等線程不安全的類，效率更高。在并發(fā)場(chǎng)景下，常會(huì)用到ConcurrentHashMap、ArrayBlockingQueue等線程安全的容器類，雖然犧牲了一些效率，但卻得到了安全。

上面提到的線程安全容器都在java.util.concurrent包下，這個(gè)包下并發(fā)容器不少，今天全部翻出來(lái)鼓搗一下。

僅做簡(jiǎn)單介紹，后續(xù)再分別深入探索。

簡(jiǎn)介

• ConcurrentHashMap：并發(fā)版HashMap

• CopyOnWriteArrayList：并發(fā)版ArrayList

• CopyOnWriteArraySet：并發(fā)Set

• ConcurrentLinkedQueue：并發(fā)隊(duì)列(基于鏈表)

• ConcurrentLinkedDeque：并發(fā)隊(duì)列(基于雙向鏈表)

• ConcurrentSkipListMap：基于跳表的并發(fā)Map

• ConcurrentSkipListSet：基于跳表的并發(fā)Set

• ArrayBlockingQueue：阻塞隊(duì)列(基于數(shù)組)

• LinkedBlockingQueue：阻塞隊(duì)列(基于鏈表)

• LinkedBlockingDeque：阻塞隊(duì)列(基于雙向鏈表)

• PriorityBlockingQueue：線程安全的優(yōu)先隊(duì)列

• SynchronousQueue：讀寫(xiě)成對(duì)的隊(duì)列

• LinkedTransferQueue：基于鏈表的數(shù)據(jù)交換隊(duì)列

• DelayQueue：延時(shí)隊(duì)列

ConcurrentHashMap 并發(fā)版HashMap

最常見(jiàn)的并發(fā)容器之一，可以用作并發(fā)場(chǎng)景下的緩存。底層依然是哈希表，但在JAVA 8中有了不小的改變，而JAVA 7和JAVA 8都是用的比較多的版本，因此經(jīng)常會(huì)將這兩個(gè)版本的實(shí)現(xiàn)方式做一些比較（比如面試中）。

一個(gè)比較大的差異就是，JAVA 7中采用分段鎖來(lái)減少鎖的競(jìng)爭(zhēng)，JAVA 8中放棄了分段鎖，采用CAS（一種樂(lè)觀鎖），同時(shí)為了防止哈希沖突嚴(yán)重時(shí)退化成鏈表（沖突時(shí)會(huì)在該位置生成一個(gè)鏈表，哈希值相同的對(duì)象就鏈在一起），會(huì)在鏈表長(zhǎng)度達(dá)到閾值（8）后轉(zhuǎn)換成紅黑樹(shù)（比起鏈表，樹(shù)的查詢效率更穩(wěn)定）。

CopyOnWriteArrayList 并發(fā)版ArrayList

并發(fā)版ArrayList，底層結(jié)構(gòu)也是數(shù)組，和ArrayList不同之處在于：當(dāng)新增和刪除元素時(shí)會(huì)創(chuàng)建一個(gè)新的數(shù)組，在新的數(shù)組中增加或者排除指定對(duì)象，最后用新增數(shù)組替換原來(lái)的數(shù)組。

適用場(chǎng)景：由于讀操作不加鎖，寫(xiě)（增、刪、改）操作加鎖，因此適用于讀多寫(xiě)少的場(chǎng)景。

局限：由于讀的時(shí)候不會(huì)加鎖（讀的效率高，就和普通ArrayList一樣），讀取的當(dāng)前副本，因此可能讀取到臟數(shù)據(jù)。如果介意，建議不用。

看看源碼感受下：

public class CopyOnWriteArrayList<E>
    implements List<E>, RandomAccess, Cloneable, java.io.Serializable {
    final transient ReentrantLock lock = new ReentrantLock();
    private transient volatile Object[] array;

    // 添加元素，有鎖
    public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock(); // 修改時(shí)加鎖，保證并發(fā)安全
        try {
            Object[] elements = getArray(); // 當(dāng)前數(shù)組
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1); // 創(chuàng)建一個(gè)新數(shù)組，比老的大一個(gè)空間
            newElements[len] = e; // 要添加的元素放進(jìn)新數(shù)組
            setArray(newElements); // 用新數(shù)組替換原來(lái)的數(shù)組
            return true;
        } finally {
            lock.unlock(); // 解鎖
        }
    }

    // 讀元素，不加鎖，因此可能讀取到舊數(shù)據(jù)
    public E get(int index) {
        return get(getArray(), index);
    }
}

CopyOnWriteArraySet 并發(fā)Set

基于CopyOnWriteArrayList實(shí)現(xiàn)（內(nèi)含一個(gè)CopyOnWriteArrayList成員變量），也就是說(shuō)底層是一個(gè)數(shù)組，意味著每次add都要遍歷整個(gè)集合才能知道是否存在，不存在時(shí)需要插入（加鎖）。

適用場(chǎng)景：在CopyOnWriteArrayList適用場(chǎng)景下加一個(gè)，集合別太大（全部遍歷傷不起）。

ConcurrentLinkedQueue 并發(fā)隊(duì)列(基于鏈表)

基于鏈表實(shí)現(xiàn)的并發(fā)隊(duì)列，使用樂(lè)觀鎖(CAS)保證線程安全。因?yàn)閿?shù)據(jù)結(jié)構(gòu)是鏈表，所以理論上是沒(méi)有隊(duì)列大小限制的，也就是說(shuō)添加數(shù)據(jù)一定能成功。

ConcurrentLinkedDeque 并發(fā)隊(duì)列(基于雙向鏈表)

基于雙向鏈表實(shí)現(xiàn)的并發(fā)隊(duì)列，可以分別對(duì)頭尾進(jìn)行操作，因此除了先進(jìn)先出(FIFO)，也可以先進(jìn)后出（FILO），當(dāng)然先進(jìn)后出的話應(yīng)該叫它棧了。

ConcurrentSkipListMap 基于跳表的并發(fā)Map

SkipList即跳表，跳表是一種空間換時(shí)間的數(shù)據(jù)結(jié)構(gòu)，通過(guò)冗余數(shù)據(jù)，將鏈表一層一層索引，達(dá)到類似二分查找的效果

ConcurrentSkipListSet 基于跳表的并發(fā)Set

類似HashSet和HashMap的關(guān)系，ConcurrentSkipListSet里面就是一個(gè)ConcurrentSkipListMap，就不細(xì)說(shuō)了。

ArrayBlockingQueue 阻塞隊(duì)列(基于數(shù)組)

基于數(shù)組實(shí)現(xiàn)的可阻塞隊(duì)列，構(gòu)造時(shí)必須制定數(shù)組大小，往里面放東西時(shí)如果數(shù)組滿了便會(huì)阻塞直到有位置（也支持直接返回和超時(shí)等待），通過(guò)一個(gè)鎖ReentrantLock保證線程安全。

用offer操作舉個(gè)例子：

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    /**
     * 讀寫(xiě)共用此鎖，線程間通過(guò)下面兩個(gè)Condition通信
     * 這兩個(gè)Condition和lock有緊密聯(lián)系（就是lock的方法生成的）
     * 類似Object的wait/notify
     */
    final ReentrantLock lock;

    /** 隊(duì)列不為空的信號(hào)，取數(shù)據(jù)的線程需要關(guān)注 */
    private final Condition notEmpty;

    /** 隊(duì)列沒(méi)滿的信號(hào)，寫(xiě)數(shù)據(jù)的線程需要關(guān)注 */
    private final Condition notFull;

    // 一直阻塞直到有東西可以拿出來(lái)
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

    // 在尾部插入一個(gè)元素，隊(duì)列已滿時(shí)等待指定時(shí)間，如果還是不能插入則返回
    public boolean offer(E e, long timeout, TimeUnit unit)
        throws InterruptedException {
        checkNotNull(e);
        long nanos = unit.toNanos(timeout);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly(); // 鎖住
        try {
            // 循環(huán)等待直到隊(duì)列有空閑
            while (count == items.length) {
                if (nanos <= 0)
                    return false;// 等待超時(shí)，返回
                // 暫時(shí)放出鎖，等待一段時(shí)間（可能被提前喚醒并搶到鎖，所以需要循環(huán)判斷條件）
                // 這段時(shí)間可能其他線程取走了元素，這樣就有機(jī)會(huì)插入了
                nanos = notFull.awaitNanos(nanos);
            }
            enqueue(e);//插入一個(gè)元素
            return true;
        } finally {
            lock.unlock(); //解鎖
        }
    }

乍一看會(huì)有點(diǎn)疑惑，讀和寫(xiě)都是同一個(gè)鎖，那要是空的時(shí)候正好一個(gè)讀線程來(lái)了不會(huì)一直阻塞嗎？

答案就在notEmpty、notFull里，這兩個(gè)出自lock的小東西讓鎖有了類似synchronized + wait + notify的功能。傳送門(mén) → 終于搞懂了sleep/wait/notify/notifyAll

LinkedBlockingQueue 阻塞隊(duì)列(基于鏈表)

基于鏈表實(shí)現(xiàn)的阻塞隊(duì)列，想比與不阻塞的ConcurrentLinkedQueue，它多了一個(gè)容量限制，如果不設(shè)置默認(rèn)為int最大值。

LinkedBlockingDeque 阻塞隊(duì)列(基于雙向鏈表)

類似LinkedBlockingQueue，但提供了雙向鏈表特有的操作。

PriorityBlockingQueue 線程安全的優(yōu)先隊(duì)列

構(gòu)造時(shí)可以傳入一個(gè)比較器，可以看做放進(jìn)去的元素會(huì)被排序，然后讀取的時(shí)候按順序消費(fèi)。某些低優(yōu)先級(jí)的元素可能長(zhǎng)期無(wú)法被消費(fèi)，因?yàn)椴粩嘤懈邇?yōu)先級(jí)的元素進(jìn)來(lái)。

SynchronousQueue 數(shù)據(jù)同步交換的隊(duì)列

一個(gè)虛假的隊(duì)列，因?yàn)樗鼘?shí)際上沒(méi)有真正用于存儲(chǔ)元素的空間，每個(gè)插入操作都必須有對(duì)應(yīng)的取出操作，沒(méi)取出時(shí)無(wú)法繼續(xù)放入。

一個(gè)簡(jiǎn)單的例子感受一下：

import java.util.concurrent.*;

public class Main {

    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        new Thread(() -> {
            try {
                // 沒(méi)有休息，瘋狂寫(xiě)入
                for (int i = 0; ; i++) {
                    System.out.println('放入: ' + i);
                    queue.put(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }).start();
        new Thread(() -> {
            try {
                // 咸魚(yú)模式取數(shù)據(jù)
                while (true) {
                    System.out.println('取出: ' + queue.take());
                    Thread.sleep((long) (Math.random() * 2000));
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }).start();
    }

}

/* 輸出:

放入: 0
取出: 0
放入: 1
取出: 1
放入: 2
取出: 2
放入: 3
取出: 3

*/

可以看到，寫(xiě)入的線程沒(méi)有任何sleep，可以說(shuō)是全力往隊(duì)列放東西，而讀取的線程又很不積極，讀一個(gè)又sleep一會(huì)。輸出的結(jié)果卻是讀寫(xiě)操作成對(duì)出現(xiàn)。

JAVA中一個(gè)使用場(chǎng)景就是Executors.newCachedThreadPool()，創(chuàng)建一個(gè)緩存線程池。

public static ExecutorService newCachedThreadPool() {
    return new ThreadPoolExecutor(
        0, // 核心線程為0，沒(méi)用的線程都被無(wú)情拋棄
        Integer.MAX_VALUE, // 最大線程數(shù)理論上是無(wú)限了，還沒(méi)到這個(gè)值機(jī)器資源就被掏空了
        60L, TimeUnit.SECONDS, // 閑置線程60秒后銷毀
        new SynchronousQueue<Runnable>()); // offer時(shí)如果沒(méi)有空閑線程取出任務(wù)，則會(huì)失敗，線程池就會(huì)新建一個(gè)線程
}

LinkedTransferQueue 基于鏈表的數(shù)據(jù)交換隊(duì)列

實(shí)現(xiàn)了接口TransferQueue，通過(guò)transfer方法放入元素時(shí)，如果發(fā)現(xiàn)有線程在阻塞在取元素，會(huì)直接把這個(gè)元素給等待線程。如果沒(méi)有人等著消費(fèi)，那么會(huì)把這個(gè)元素放到隊(duì)列尾部，并且此方法阻塞直到有人讀取這個(gè)元素。和SynchronousQueue有點(diǎn)像，但比它更強(qiáng)大。

DelayQueue 延時(shí)隊(duì)列

可以使放入隊(duì)列的元素在指定的延時(shí)后才被消費(fèi)者取出，元素需要實(shí)現(xiàn)Delayed接口。

總結(jié)

上面簡(jiǎn)單介紹了JAVA并發(fā)包下的一些容器類，知道有這些東西，遇到合適的場(chǎng)景時(shí)就能想起有個(gè)現(xiàn)成的東西可以用了。想要知其所以然，后續(xù)還得再深入探索一番。