多線程與高併發09-併發容器（二）

時間  2020-05-12
標籤 多線程 併發 容器 欄目 Java 简体版
原文   原文鏈接

ConcurrentSkipList系列

  • ConcurrentSkipListMap：跳錶實現有序Map
  • ConcurrentSkipListSet：跳錶實現有序Set
TreeMap和TreeSet使用 紅黑樹按照key的順序（天然順序、自定義順序）來使得鍵值對有序存儲，可是隻能在單線程下安全使用；多線程下想要使鍵值對按照key的順序來存儲，則須要使用ConcurrentSkipListMap和ConcurrentSkipListSet，分別用以代替TreeMap和TreeSet，存入的數據按key排序。在實現上，ConcurrentSkipListSet 本質上就是ConcurrentSkipListMap

image.png

瞭解什麼是SkipList

二分查找和AVL樹查找

連續數組的侷限：二分查找要求元素能夠隨機訪問，因此決定了須要把元素存儲在連續內存。這樣查找確實很快，可是插入和刪除元素的時候，爲了保證元素的有序性，就須要大量的移動元素了前端

二叉查找樹：若是須要的是一個可以進行二分查找，又能快速添加和刪除元素的數據結構，首先就是二叉查找樹，二叉查找樹在最壞狀況下可能變成一個鏈表java

平衡二叉樹：出現了平衡二叉樹，根據平衡算法的不一樣有AVL樹，B-Tree，B+Tree，紅黑樹等，可是AVL樹實現起來比較複雜，平衡操做較難理解，這時候就能夠用SkipList跳躍表結構算法

什麼是跳錶

傳統意義的單鏈表是一個線性結構，向有序的鏈表中插入一個節點須要O(n)的時間，查找操做須要O(n)的時間編程

image.png

若是咱們使用上圖所示的跳躍表，就能夠減小查找所需時間爲O(n/2)，由於咱們能夠先經過每一個節點的最上面的指針先進行查找，這樣子就能跳過一半的節點後端

好比咱們想查找50，首先和20比較，大於20以後，在和40進行比較，而後在和70進行比較，發現70大於50，說明查找的點在40和50之間，從這個過程當中，咱們能夠看出，查找的時候跳過了30數組

跳躍表其實也是一種經過「空間來換取時間」的一個算法，令鏈表的每一個結點不只記錄next結點位置，還能夠按照level層級分別記錄後繼第level個結點。此法使用的就是「先大步查找肯定範圍，再逐漸縮小迫近」的思想進行的查找。跳躍表在算法效率上很接近紅黑樹緩存

跳躍表又被稱爲機率，或者說是隨機化的數據結構，目前開源軟件 Redis 和 lucence都有用到它安全

都是線程安全的Map實現，ConcurrentHashMap的性能和存儲空間要優於ConcurrentSkipListMap，可是ConcurrentSkipListMap有一個功能：它會按照鍵的順序進行排序數據結構

ConcurrentLinkedQueue

無界非阻塞隊列：它是一個基於鏈表的無界線程安全隊列。該隊列的元素遵循先進先出的原則。頭是最早加入的，尾是最近加入的。插入元素是追加到尾上。提取一個元素是從頭提取多線程

你們能夠當作是LinkedList的併發版本，經常使用方法：

  • concurrentLinkedQueue.add("c")
  • concurrentLinkedQueue.offer("d")：將指定元素插入到此隊列的尾部
  • concurrentLinkedQueue.peek()：檢索並不移除此隊列的頭，若是此隊列爲空，則返回 null
  • concurrentLinkedQueue.poll()：檢索並移除此隊列的頭，若是此隊列爲空，則返回 null

寫時複製容器

什麼是寫時複製容器

CopyOnWriteArrayList和CopyOnWriteArraySet

CopyOnWrite：容器即寫時複製的容器。通俗的理解是當咱們往一個容器添加元素的時候，不直接往當前容器添加，而是先將當前容器進行Copy，複製出一個新的容器，而後新的容器裏添加元素，添加完元素以後，再將原容器的引用指向新的容器

這樣作的好處是咱們能夠對CopyOnWrite容器進行併發的讀，而不須要加鎖，由於當前容器不會添加任何元素。因此CopyOnWrite容器也是一種讀寫分離的思想，讀和寫不一樣的容器。若是讀的時候有多個線程正在向CopyOnWriteArrayList添加數據，讀仍是會讀到舊的數據，由於寫的時候不會鎖住舊的CopyOnWriteArrayList

CopyOnWrite併發容器用於對於（讀多寫少）絕大部分訪問都是讀，且只是偶爾寫的併發場景。好比白名單，黑名單，商品類目的訪問和更新場景，假如咱們有一個搜索網站，用戶在這個網站的搜索框中，輸入關鍵字搜索內容，可是某些關鍵字不容許被搜索。這些不能被搜索的關鍵字會被放在一個黑名單當中，黑名單天天晚上更新一次。當用戶搜索時，會檢查當前關鍵字在不在黑名單當中，若是在，則提示不能搜索

使用CopyOnWriteMap須要注意兩件事情：

  • 減小擴容開銷。根據實際須要，初始化CopyOnWriteMap的大小，避免寫時擴容的開銷
  • 使用批量添加。由於每次添加，容器每次都會進行復制，因此減小添加次數，能夠減小容器的複製次數

寫時複製容器的問題

性能問題

每次修改都建立一個新數組，而後複製全部內容，若是數組比較大，修改操做又比較頻繁，能夠想象，性能是很低的，並且內存開銷會很大

數據一致性問題

CopyOnWrite容器只能保證數據的最終一致性，不能保證數據的實時一致性。因此若是你但願寫入的的數據，立刻能讀到，不要使用CopyOnWrite容器

阻塞隊列BlockingQueue

隊列

image.png

隊列是一種特殊的線性表，特殊之處在於它只容許在表的前端（front）進行刪除操做，而在表的後端（rear）進行插入操做，和棧同樣，隊列是一種操做受限制的線性表。進行插入操做的端稱爲隊尾，進行刪除操做的端稱爲隊頭

在隊列中插入一個隊列元素稱爲入隊，從隊列中刪除一個隊列元素稱爲出隊。由於隊列只容許在一端插入，在另外一端刪除，因此只有最先進入隊列的元素才能最早從隊列中刪除，故隊列又稱爲先進先出（FIFO—first in first out）線性表

什麼是阻塞隊列

  • 支持阻塞的插入方法：意思是當隊列滿時，隊列會阻塞插入元素的線程，直到隊列不滿。
  • 支持阻塞的移除方法：意思是在隊列爲空時，獲取元素的線程會等待隊列變爲非空。

在併發編程中使用生產者和消費者模式可以解決絕大多數併發問題。該模式經過平衡生產線程和消費線程的工做能力來提升程序總體處理數據的速度

在線程世界裏，生產者就是生產數據的線程，消費者就是消費數據的線程。在多線程開發中，若是生產者處理速度很快，而消費者處理速度很慢，那麼生產者就必須等待消費者處理完，才能繼續生產數據。一樣的道理，若是消費者的處理能力大於生產者，那麼消費者就必須等待生產者

爲了解決這種生產消費能力不均衡的問題，便有了生產者和消費者模式。生產者和消費者模式是經過一個容器來解決生產者和消費者的強耦合問題。生產者和消費者彼此之間不直接通訊，而是經過阻塞隊列來進行通訊，因此生產者生產完數據以後不用等待消費者處理，直接扔給阻塞隊列，消費者不找生產者要數據，而是直接從阻塞隊列裏取，阻塞隊列就至關於一個緩衝區，平衡了生產者和消費者的處理能力

阻塞隊列經常使用於生產者和消費者的場景，生產者是向隊列裏添加元素的線程，消費者是從隊列裏取元素的線程。阻塞隊列就是生產者用來存放元素、消費者用來獲取元素的容器
image.png

  • 拋出異常：當隊列滿時，若是再往隊列裏插入元素，會拋出IllegalStateException（"Queuefull"）異常。當隊列空時，從隊列裏獲取元素會拋出NoSuchElementException異常
  • 返回特殊值：當往隊列插入元素時，會返回元素是否插入成功，成功返回true。若是是移除方法，則是從隊列裏取出一個元素，若是沒有則返回null
  • 一直阻塞：當阻塞隊列滿時，若是生產者線程往隊列裏put元素，隊列會一直阻塞生產者線程，直到隊列可用或者響應中斷退出。當隊列空時，若是消費者線程從隊列裏take元素，隊列會阻塞住消費者線程，直到隊列不爲空
  • 超時退出：當阻塞隊列滿時，若是生產者線程往隊列裏插入元素，隊列會阻塞生產者線程一段時間，若是超過了指定的時間，生產者線程就會退出

經常使用阻塞隊列

  • ArrayBlockingQueue：一個由數組結構組成的有界阻塞隊列
  • LinkedBlockingQueue：一個由鏈表結構組成的無界阻塞隊列
  • PriorityBlockingQueue：一個支持優先級排序的無界阻塞隊列
  • DelayQueue：一個使用優先級隊列實現執行定時任務的無界阻塞隊列
  • SynchronousQueue：一個不存儲元素的阻塞隊列
  • LinkedTransferQueue：一個由鏈表結構組成的無界阻塞隊列

LinkedBlockingDeque：一個由鏈表結構組成的雙向阻塞隊列。

以上的阻塞隊列都實現了BlockingQueue接口，也都是線程安全的

有界與無界

有限隊列就是長度有限，滿了之後生產者會阻塞，無界隊列就是裏面能放無數的東西而不會由於隊列長度限制被阻塞，固然空間限制來源於系統資源的限制，若是處理不及時，致使隊列愈來愈大愈來愈大，超出必定的限制導致內存超限，操做系統或者JVM幫你解決煩惱，直接把你 OOM kill 省事了

無界也會阻塞，爲什麼？

由於阻塞不只僅體如今生產者放入元素時會阻塞，消費者拿取元素時，若是沒有元素，一樣也會阻塞

ArrayBlockingQueue

是一個用數組實現的有界阻塞隊列。此隊列按照先進先出（FIFO）的原則對元素進行排序。默認狀況下不保證線程公平的訪問隊列，所謂公平訪問隊列是指阻塞的線程，能夠按照阻塞的前後順序訪問隊列，即先阻塞線程先訪問隊列。非公平性是對先等待的線程是非公平的，當隊列可用時，阻塞的線程均可以爭奪訪問隊列的資格，有可能先阻塞的線程最後才訪問隊列。初始化時有參數能夠設置
示例代碼：

import java.util.Random;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

public class T06_ArrayBlockingQueue {

    static BlockingQueue<String> strs = new ArrayBlockingQueue<>(10);

    static Random r = new Random();

    public static void main(String[] args) throws InterruptedException {
        for (int i = 0; i < 10; i++) {
            strs.put("a" + i);
        }
        
        strs.put("aaa"); //滿了就會等待，程序阻塞
        strs.add("aaa"); //滿了就會拋出異常 full queue
        strs.offer("aaa"); //滿了會當即返回false
        strs.offer("aaa", 10, TimeUnit.SECONDS); //滿了等待10秒，不成功返回false
        System.out.println(strs);
    }
}

LinkedBlockingQueue

是一個用鏈表實現的無界阻塞隊列。此隊列的默認和最大長度爲Integer.MAX_VALUE。此隊列按照先進先出的原則對元素進行排序
示例代碼：

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

public class T05_LinkedBlockingQueue {

    static BlockingQueue<String> strs = new LinkedBlockingQueue<>();

    static Random r = new Random();

    public static void main(String[] args) {
        new Thread(() -> {
            for (int i = 0; i < 10; i++) {
                try {
                    strs.put("a" + i); //若是滿了，就會等待
                    TimeUnit.MILLISECONDS.sleep(r.nextInt(1000));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }, "p1").start();

        for (int i = 0; i < 5; i++) {
            new Thread(() -> {
                for (;;) {
                    try {
                        System.out.println(Thread.currentThread().getName() + " take -" + strs.take()); //若是空了，就會等待
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }, "c" + i).start();

        }
    }
}

運行結果：

c0 take -a0
c1 take -a1
c2 take -a2
c3 take -a3
c4 take -a4
c0 take -a5
c1 take -a6
c2 take -a7
c3 take -a8
c4 take -a9

Array實現和Linked實現的區別

隊列中鎖的實現不一樣

ArrayBlockingQueue實現的隊列中的鎖是沒有分離的，即生產和消費用的是同一個鎖

LinkedBlockingQueue實現的隊列中的鎖是分離的，即生產用的是putLock，消費是takeLock

在生產或消費時操做不一樣

ArrayBlockingQueue實現的隊列中在生產和消費的時候，是直接將對象插入或移除的

LinkedBlockingQueue實現的隊列中在生產和消費的時候，須要把對象轉換爲Node<E>進行插入或移除(由於須要增長鏈表指針)，會影響性能

隊列大小初始化方式不一樣

ArrayBlockingQueue實現的隊列中必須指定隊列的大小

LinkedBlockingQueue實現的隊列中能夠不指定隊列的大小，可是默認是Integer.MAX_VALUE

PriorityBlockingQueue

PriorityBlockingQueue是一個支持優先級的無界阻塞隊列。默認狀況下元素採起天然順序升序排列。也能夠自定義類實現compareTo()方法來指定元素排序規則，或者初始化PriorityBlockingQueue時，指定構造參數Comparator來對元素進行排序。須要注意的是不能保證同優先級元素的順序
示例代碼：

import java.util.PriorityQueue;

public class T07_01_PriorityQueque {
    public static void main(String[] args) {
        PriorityQueue<String> q = new PriorityQueue<>();

        q.add("c");
        q.add("e");
        q.add("a");
        q.add("d");
        q.add("z");

        for (int i = 0; i < 5; i++) {
            System.out.println(q.poll());
        }

    }
}

運行結果：

a
c
d
e
z

DelayQueue

是一個支持延時獲取元素的無界阻塞隊列。隊列使用PriorityQueue來實現。隊列中的元素必須實現Delayed接口，在建立元素時能夠指定多久才能從隊列中獲取當前元素。只有在延遲期滿時才能從隊列中提取元素

DelayQueue很是有用，能夠將DelayQueue運用在如下應用場景：

緩存系統的設計能夠用DelayQueue保存緩存元素的有效期，使用一個線程循環查詢DelayQueue，一旦能從DelayQueue中獲取元素時，表示緩存有效期到了。還有訂單到期，限時支付等等
示例代碼：

import java.util.Random;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

public class T07_DelayQueue {

    static BlockingQueue<MyTask> tasks = new DelayQueue<>();

    static Random r = new Random();
    
    static class MyTask implements Delayed {
        String name;
        long runningTime;
        
        MyTask(String name, long rt) {
            this.name = name;
            this.runningTime = rt;
        }

        @Override
        public int compareTo(Delayed o) {
            if(this.getDelay(TimeUnit.MILLISECONDS) < o.getDelay(TimeUnit.MILLISECONDS))
                return -1;
            else if(this.getDelay(TimeUnit.MILLISECONDS) > o.getDelay(TimeUnit.MILLISECONDS)) 
                return 1;
            else 
                return 0;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            
            return unit.convert(runningTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
        
        
        @Override
        public String toString() {
            return name + " " + runningTime;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        long now = System.currentTimeMillis();
        MyTask t1 = new MyTask("t1", now + 1000);
        MyTask t2 = new MyTask("t2", now + 2000);
        MyTask t3 = new MyTask("t3", now + 1500);
        MyTask t4 = new MyTask("t4", now + 2500);
        MyTask t5 = new MyTask("t5", now + 500);
        
        tasks.put(t1);
        tasks.put(t2);
        tasks.put(t3);
        tasks.put(t4);
        tasks.put(t5);
        
        System.out.println(tasks);
        
        for(int i=0; i<5; i++) {
            System.out.println(tasks.take());
        }
    }
}

運行結果：

[t5 1586608841726, t1 1586608842226, t3 1586608842726, t4 1586608843726, t2 1586608843226]
t5 1586608841726
t1 1586608842226
t3 1586608842726
t2 1586608843226
t4 1586608843726

SynchronousQueue

是一個不存儲元素的阻塞隊列。每個put操做必須等待一個take操做，不然不能繼續添加元素。SynchronousQueue能夠當作是一個傳球手，負責把生產者線程處理的數據直接傳遞給消費者線程。隊列自己並不存儲任何元素，很是適合傳遞性場景。SynchronousQueue的吞吐量高於LinkedBlockingQueue和ArrayBlockingQueue
示例代碼：

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.SynchronousQueue;

public class T08_SynchronusQueue { //容量爲0
    public static void main(String[] args) throws InterruptedException {
        BlockingQueue<String> strs = new SynchronousQueue<>();
        
        new Thread(()->{
            try {
                System.out.println(strs.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();

        strs.put("aaa"); //阻塞等待消費者消費
        //strs.put("bbb");
        //strs.add("aaa");
        System.out.println(strs.size());
    }
}

運行結果：

aaa
0

LinkedTransferQueue

多了tryTransfer和transfer方法

  • transfer方法

若是當前有消費者正在等待接收元素（消費者使用take()方法或帶時間限制的poll()方法時），transfer方法能夠把生產者傳入的元素馬上transfer（傳輸）給消費者。若是沒有消費者在等待接收元素，transfer方法會將元素存放在隊列的tail節點，並等到該元素被消費者消費了才返回

  • tryTransfer方法

tryTransfer方法是用來試探生產者傳入的元素是否能直接傳給消費者。若是沒有消費者等待接收元素，則返回false。和transfer方法的區別是tryTransfer方法不管消費者是否接收，方法當即返回，而transfer方法是必須等到消費者消費了才返回
示例代碼：

package com.mashibing.juc.c_025;

import java.util.concurrent.LinkedTransferQueue;

public class T09_TransferQueue {
    public static void main(String[] args) throws InterruptedException {
        LinkedTransferQueue<String> strs = new LinkedTransferQueue<>();
        
        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() +":"+ strs.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();
        
        strs.transfer("aaa");
        
        strs.put("bbb");


        new Thread(() -> {
            try {
                System.out.println(Thread.currentThread().getName() +":"+strs.take());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }).start();


    }
}

運行結果：

Thread-0:aaa
Thread-1:aaa

LinkedBlockingDeque

LinkedBlockingDeque是一個由鏈表結構組成的雙向阻塞隊列。所謂雙向隊列指的是能夠從隊列的兩端插入和移出元素。雙向隊列由於多了一個操做隊列的入口，在多線程同時入隊時，也就減小了一半的競爭

多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法，以First單詞結尾的方法，表示插入、獲取（peek）或移除雙端隊列的第一個元素。以Last單詞結尾的方法，表示插入、獲取或移除雙端隊列的最後一個元素。另外，插入方法add等同於addLast，移除方法remove等效於removeFirst。可是take方法卻等同於takeFirst，不知道是否是JDK的bug，使用時仍是用帶有First和Last後綴的方法更清楚。在初始化LinkedBlockingDeque時能夠設置容量防止其過分膨脹。另外，雙向阻塞隊列能夠運用在「工做竊取」(ForkJoinPool)模式中

瞭解阻塞隊列的實現原理

使用了wait/notify模式實現。所謂通知模式，就是當生產者往滿的隊列裏添加元素時會阻塞住生產者，當消費者消費了一個隊列中的元素後，會通知生產者當前隊列可用。經過查看JDK源碼發現ArrayBlockingQueue使用了Condition來實現。其他隊列的實現，你們能夠自行查看，隊列的實現的代碼整體來講，並不複雜

相關文章
相關標籤/搜索
每日一句
    每一个你不满意的现在，都有一个你没有努力的曾经。
本站公眾號
   歡迎關注本站公眾號,獲取更多信息
聯繫我們 最近搜索 最新文章 沪ICP备13005482号-10 MyBatis教程 SQL 教程 MySQL教程 Java 教程 Thymeleaf 教程 Hibernate教程 Spring教程 Redis教程