并發編程核心——九、阻塞佇列

使用LOCK.notify,LOCK.wait模擬一個阻塞佇列.

阻塞佇列也是一個佇列，擁有佇列某個時刻擁有元素數量，佇列最大容納元素數量，佇列最小數量，佇列這些屬性，
阻塞的含義是如果佇列中元素個數與佇列最大容納元素數量相同，那么再放元素，放元素的執行緒會進入wait佇列的阻塞狀態，等待其他執行緒喚醒后才能放元素；如果佇列中的元素個數與最小容納元素數量相同，那么取資料的執行緒會進入wait佇列的阻塞狀態，等待其他執行緒喚醒后才能取資料，
增加屬性：鎖
操作：構造方法，私有屬性的公有獲取方法，獲取隊首元素的方法（加鎖），獲取隊尾元素資料（加鎖）

代碼1：自己封裝的阻塞佇列

package com.bjsxt.chapter09;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 模擬阻塞佇列
 * @author wmr
 * @date 2021/2/20
 */
public class MyQueue {
    // 佇列
    private LinkedList<Object> list = new LinkedList<>();
    // 元素個數
    private AtomicLong count = new AtomicLong(0);
    // 最大數量
    private final long maxSize;
    // 最小數量
    private final long minSize = 0;
    // 鎖
    private final Object LOCK = new Object();
    // 構造方法
    public MyQueue(long maxSize) {
        this.maxSize = maxSize;
    }
    // 獲取元素個數(封裝變數的存取操作到方法中，這樣便于維護變數的存取增加的操作)
    public long getCount(){
        return this.count.get();
    }
    // 獲取佇列最多能放元素數量
    public long getMaxSize(){
        return this.maxSize;
    }
    // 獲取佇列最少能放元素數量
    public long getMinSize(){
        return this.minSize;
    }
    // 入隊
    public void put(Object o){
        synchronized (LOCK){
            // 佇列滿，入隊執行緒進入wait佇列的阻塞狀態
            if(this.count.get() == this.maxSize){
                try{
                    LOCK.wait();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            // 否則，入隊，數量加一，喚醒其他執行緒執行出隊操作
            this.list.add(o);
            this.count.incrementAndGet();
            LOCK.notify();
            try{
                TimeUnit.SECONDS.sleep(2);
            }catch (InterruptedException e){
                e.printStackTrace();
            }

        }
    }
    // 出隊
    public Object take(){
        Object ret = null;
        synchronized (LOCK){
            // 佇列空，出隊執行緒進入wait佇列的阻塞狀態
            if(this.count.get() == this.minSize){
                try{
                    LOCK.wait();
                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }
            // 否則，回傳第一個元素，洗掉第一個元素，數量減一，喚醒其他執行緒進行入隊操作
            ret = this.list.getFirst();
            this.list.removeFirst();
            this.count.decrementAndGet();
            LOCK.notify();
            try{
                TimeUnit.SECONDS.sleep(2);
            }catch (InterruptedException e){
                e.printStackTrace();
            }
        }
        return ret;
    }
}

代碼2：程式員業務操作的類，測驗阻塞佇列的使用

package com.bjsxt.chapter09;


import java.util.concurrent.TimeUnit;

/**
 * @author wmr
 * @date 2021/2/20
 */
public class Demo {
    public static void main(String[] args) throws InterruptedException {
        System.out.println("-----------main------------");
        // 模擬程式員在外面業務代碼中使用阻塞佇列，呼叫構造器創建一個阻塞佇列
        final MyQueue myQueue = new MyQueue(5);
        // 向阻塞佇列中依次放進"a","b","c","d","e"五個元素
        myQueue.put("a");
        System.out.println("元素" + "a" + "入隊");
        myQueue.put("b");
        System.out.println("元素" + "b" + "入隊");
        myQueue.put("c");
        System.out.println("元素" + "c" + "入隊");
        myQueue.put("d");
        System.out.println("元素" + "d" + "入隊");
        myQueue.put("e");
        System.out.println("元素" + "e" + "入隊");

        System.out.println("此刻佇列中元素個數："+myQueue.getCount());
        System.out.println("佇列最多能放元素數量: "+myQueue.getMaxSize());
        System.out.println("佇列最少元素數量: "+myQueue.getMinSize());

        // 創建一個放資料的執行緒
        Thread putThread = new Thread(() -> {
            myQueue.put("f");
            System.out.println("元素" + "f" + "入隊");
            myQueue.put("g");
            System.out.println("元素" + "g" + "入隊");
        },"put-Thread");

        // 創建一個取資料的執行緒
        Thread takeThread = new Thread(() -> {
            Object o = myQueue.take();
            System.out.println("元素"+o+"出隊");
            o = myQueue.take();
            System.out.println("元素"+o+"出隊");
        },"take-Thread");

        // 啟動放資料的執行緒，因為阻塞佇列滿了，所以放資料執行緒放不進去f,g，進入wait佇列的阻塞狀態等待被LOCK.notify喚醒
        System.out.println("-----------put-Thread------------");
        putThread.start();

        // main執行緒進入普通的阻塞狀態2秒，使得放資料的執行緒putThread一定被執行
        TimeUnit.SECONDS.sleep(2);

        // 啟動取資料的執行緒，取出一個資料后會喚醒進入wait佇列的putThread，使其進入鎖池的阻塞狀態，與其他執行緒共同競爭cpu
        System.out.println("-----------take-Thread------------");
        takeThread.start();
    }
}

運行結果：如圖所示，阻塞佇列可用，
分析：
總共有3個執行緒，main、put-Thread、take-Thread，main執行緒創建一個最多容納5個元素的阻塞佇列，向阻塞佇列依次入隊5個元素，put-Thread入隊2個元素，take-Thread出隊2個元素，對于每一個具體的執行緒來說，執行順序沒有問題，
對于出隊，入隊執行緒的通信來說，呼叫入隊執行緒的start并沒有入隊資訊列印說明入隊執行緒阻塞了，之后出隊執行緒將隊首元素出隊，喚醒入隊執行緒，入隊執行緒蔣f入隊，之后又進入wait狀態，只能進行出隊操作，這與腦海中的分析相符，證明完成了阻塞佇列的封裝，

備注：

System.out.println(“元素” + “g” + “入隊”);在入隊操作里面輸出，會出現列印前后不一致現象：
System.out.println(“元素” + “f” + “入隊”);
System.out.println(“元素” + “a” + “出隊”);
所以干脆寫到外面了，封裝的阻塞佇列里面只完成核心的入隊，出隊，數量增加，減少操作，對于給自己看的顯示輸出資訊解釋資訊放在外面，

另外就是對于佇列屬性的獲取，使用方法封裝了，在封裝的類中可以訪問自己的私有變數，如果是在其他類中呼叫必須通過public的獲取方法獲取變數的值，如果直接使用public修飾變數，不知道情況的人可以直接設定值，佇列最小數量變成-1就不好了！！如果被其他人設定成-1，那么就報錯或者將其設定為0.

BlockingQueue 即阻塞佇列，它是基于 ReentrantLock，依據它的基本原理，我們可 以實作 Web 中的長連接聊天功能，當然其最常用的還是用于實作生產者與消費者模式，大 致如下圖所示：

在 Java 中，BlockingQueue 是一個介面，它的實作類有 ArrayBlockingQueue、 DelayQueue 、 LinkedBlockingDeque 、 LinkedBlockingQueue 、 PriorityBlockingQueue、SynchronousQueue 等，它們的區別主要體現在存盤結構上或 對元素操作上的不同，但是對于 take 與 put 操作的原理，卻是類似的，

put(an Object)：把 an Object 加到 BlockingQueue 里，如果 BlockQueue 沒有 空間，則呼叫此方法的執行緒被阻斷，直到 BlockingQueue 里面有空間再繼續，

take：取走 BlockingQueue 里排在首位的物件，若 BlockingQueue 為空，阻斷進入 等待狀態直到 BlockingQueue 有新的資料被加入