Java常用的并發(fā)容器

這篇文章主要講解了“Java常用的并發(fā)容器”，文中的講解內(nèi)容簡單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來研究和學(xué)習(xí)“Java常用的并發(fā)容器”吧！

并發(fā)容器介紹

1. ConcurrentHashMap：并發(fā)版HashMap

2. CopyOnWriteArrayList：并發(fā)版ArrayList

3. CopyOnWriteArraySet：并發(fā)Set

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

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

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

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

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

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

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

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

12. SynchronousQueue：讀寫成對(duì)的隊(duì)列

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

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

1.ConcurrentHashMap 并發(fā)版HashMap

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

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

2.CopyOnWriteArrayList 并發(fā)版ArrayList

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

適用場景：由于讀操作不加鎖，寫（增、刪、改）操作加鎖，因此適用于讀多寫少的場景。

局限：由于讀的時(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ù)組替換原來的數(shù)組
            return true;
        } finally {
            lock.unlock(); // 解鎖
        }
    }
    // 讀元素，不加鎖，因此可能讀取到舊數(shù)據(jù)
    public E get(int index) {
        return get(getArray(), index);
    }
}

3.CopyOnWriteArraySet 并發(fā)Set

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

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

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

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

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

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

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

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

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

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

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

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

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

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
    /**
     * 讀寫共用此鎖，線程間通過下面兩個(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ì)列沒滿的信號(hào)，寫數(shù)據(jù)的線程需要關(guān)注 */
    private final Condition notFull;
    // 一直阻塞直到有東西可以拿出來
    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)疑惑，讀和寫都是同一個(gè)鎖，那要是空的時(shí)候正好一個(gè)讀線程來了不會(huì)一直阻塞嗎？

答案就在notEmpty、notFull里，這兩個(gè)出自lock的小東西讓鎖有了類似synchronized + wait + notify的功能。

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

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

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

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

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

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

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

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

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

import java.util.concurrent.*;
public class Main {
    public static void main(String[] args) {
        SynchronousQueue<Integer> queue = new SynchronousQueue<>();
        new Thread(() -> {
            try {
                // 沒有休息，瘋狂寫入
                for (int i = 0; ; i++) {
                    System.out.println("放入: " + i);
                    queue.put(i);
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        new Thread(() -> {
            try {
                // 咸魚模式取數(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
*/

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

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

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

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

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

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

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

感謝各位的閱讀，以上就是“Java常用的并發(fā)容器”的內(nèi)容了，經(jīng)過本文的學(xué)習(xí)后，相信大家對(duì)Java常用的并發(fā)容器這一問題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是億速云，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！