高效使用阻塞隊列：生產者-消費者模型的最佳實踐 | 阻塞隊列 | 3 種實現

需要有一個橋樑將生產者、消費者連接起來，這個橋樑就是「隊列」，這樣兩種才可以通訊（相互傳遞是否已經處理完畢，或是生產出需要處理的物件）

● 為什麼隊列？不是其他數據結構

因為隊列（Queue）這種數據結構本身就具有 FIFO 的特性，這個特性擁有「排隊」的功能，能夠「有順序性」的處理事件

阻塞意味著「等待」，因為不管是生產者還是消費者，雙方往往不會有資源對等的情況，更常碰到的是其中一方缺乏資源，而另外一方不斷搜尋（Pulling）的情況

這意味著資源的耗費，所以 生產者、消費者模型 往往需要一個可阻塞的容器

● 為什麼需要阻塞？

阻塞在執行序來說就是休眠，意味這讓出 CPU 資源，這種特性運作在隊列中代表的含義就是等待任務（並且不耗費 CPU 資源），可以大大提升應用的性能

在這裡我們再加入多執行序（Multi-Thread）的特性，我們知道多執行序在運行時沒有任何的鎖 🔒 那就是一種非安全操作，所以我們需要這個通訊容器內擁有鎖 🔒

而怎麼鎖，幾把鎖，又該如何挑選鎖… 可以點擊以下連結去了解鎖，而這篇文章底下會說明如何實現「生產者、消費者模型」

對鎖 🔒 不清楚的讀者，可以點擊 全面解析多執行緒與同步技術：SYNC、CAS、ThreadLocal | 公平鎖、可重入鎖、樂觀鎖 參考

A. ArrayBlockingQueue 隊列 : 由數據結構靜態 Queue 所組成的 「有界」 堵塞隊列

B. LinkedBlockingQueue 隊列 : 由數據結構動態 Queue 所組成的 「有界」 堵塞隊列

C. LinkedBlockingDeque 隊列 : 由數據結構雙向 Queue 所組成的 「雙向」 堵塞隊列

A. PriorityBlockingQueue 隊列 : 支持優先即排序的 「無界」 堵塞隊列

B. LinkedTransferQueue 隊列 : 由數據結構 Queue 所組成的 「無界」 堵塞隊列，實現了重要界面 TransferQueue

C. DelayQueue 隊列 : 使用優先即排序的「無界」堵塞隊列，剩餘時間越短越快取出，並且時間到後才能取出

D. SynchroniusQueue 隊列 : 不儲存元素的隊列，解偶生產者 ＆ 消費者

以下簡單的使用靜態資源（static member）

abstract class abCommonClass {
    protected static final int SIZE = 10;
    protected static ArrayBlockingQueue<String> a = new ArrayBlockingQueue<>(SIZE, true);
}

● 為什麼要使用靜態資源？

其實只是為個簡單示範而已，這樣我在外部就不需要創建隊列… 而使用靜態（static）成員的原因是因為，「生產者、消費者」的目標容器要相同，否則就無法協作！

首先消費者進入無限循環，不斷的取出元素，直到元素都被取完就「阻塞」（poll 方法的特性，如果沒有元素可取 poll 就會阻塞當前執行緒）

class abConsumer extends abCommonClass implements Runnable {

    @Override
    public void run() {
        while(true) {
            try {
                TimeUnit.SECONDS.sleep(3);
                a.poll();
                System.out.println("Poll Data " + a.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

相同，生產者進入無限循環，不斷的放入元素，直到隊列被填滿就「阻塞」（offer 方法的特性，如果隊列滿 offer 就會阻塞當前執行緒）

class abProducer extends abCommonClass implements Runnable {

    @Override
    public void run() {
        while(true) {
            try {
                TimeUnit.SECONDS.sleep(1);
                a.offer("String");
                System.out.println("Offer Data " + a.size());
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public class testArrayBlockingQueue {

    public static void main(String[] args) {
        abConsumer c = new abConsumer();
        abProducer p = new abProducer();

        new Thread(c).start();
        new Thread(p).start();
    }
}

--實做--

這邊簡單的使用靜態資源（成員 p）作為鎖

abstract class commonKey {
    // 限制 Queue 的大小
    protected static final int SIZE = 10;
    
    // Java 的阻塞隊列
    protected static PriorityQueue<String> p = new PriorityQueue<>(SIZE);
}

● Java 的阻塞隊列內就已經有鎖了，為什麼還要鎖？

這是因為阻塞隊列的鎖是針對隊列內的操作元素，而不是外部的操作，我們現在的鎖是針對 生產者、消費者操作的鎖

首先消費者開始運行後，^1.進入 while 無限循環，不斷判斷是否有元素要消耗，並且在 ^2. 判斷元素時使用鎖 🔒 保障資源的安全消耗

鎖配合 synchronized 關鍵字

如果 ^3. 「沒有元素」要消耗則執行序進入 wait() 阻塞休眠，^4. 消耗則使用 poll() 方法，並呼叫 notify 通知生產這繼續生產

class consumer extends commonKey implements Runnable {

    @Override
    public void run() {
        while(true) {
            synchronized (p) {
                while(p.size() == 0) {
                    try {
                        System.out.println("Queue Empty can not Poll");
                        p.wait();	// Object's method
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        p.notify();
                    }
                }

                try {
                    //TimeUnit.SECONDS.sleep(1);
                    System.out.println("Poll Data " + p.size());
                    p.poll();
                    p.notify();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

同樣生產者者開始運行後，^1.進入 while 無限循環，不斷判斷是否有元素需要消耗，並且在 ^2. 判斷元素時使用鎖 🔒 保障資源的安全消耗

鎖配合 synchronized 關鍵字

如果 ^3. 「元素超過隊列限制」則生產者執行序進入 wait() 阻塞休眠，^4. 生產則使用 offer() 方法，並呼叫 notify 通知消費者繼續消費

class producer extends commonKey implements Runnable {

    @Override
    public void run() {
        while(true) {
            synchronized (p) {
                while(p.size() == SIZE) {
                    try {
                         System.out.println("Queue Full can not offer anymore");
                         p.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        p.notify();
                    }
                }


                try {
                    TimeUnit.SECONDS.sleep(1);
                     System.out.println("producer " + p.size());
                    p.offer("Hello");
                    p.notify();
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

public class testBlockingQueue {

    public static void main(String[] args) {
        consumer c = new consumer();
        producer p = new producer();

        Thread t1 = new Thread(c);
        Thread t2 = new Thread(p);

        t1.start();
        t2.start();
    }

}

這邊進階改成 ReentrantLock 作為鎖，並分別出兩個條件分別給生產者（使用 con 成員）、消費者控制（使用 pro 成員）

class rCommonReentrant {
    // 隊列大小限制
    protected static final int SIZE = 10;

    // Java 的阻塞隊列
    protected static final PriorityQueue<String> p = new PriorityQueue<>(SIZE);

    // Java 的可重入鎖
    protected static ReentrantLock r = new ReentrantLock();

    // 創建兩個條件去解鎖
    protected static Condition con = r.newCondition();
    protected static Condition pro = r.newCondition();
}

這裡我們只說明不同點，首先 ^1. 使用鎖不再是透過 synchronized 關鍵字，而是使用 lock、unlock 操作，再來 ^2. 鎖的等待使用消費者自身的 Condition#await 方法

在消耗完元素後使用生產者的 Condition#signal 喚醒生產者

class rConsumer extends rCommonReentrant implements Runnable {

    @Override
    public void run() {
        while(true) {
            try {
                r.lock();

                while(p.size() == 0) {
                    System.out.println("Queue Empty can not Poll");
                    con.await();
                }

                //TimeUnit.SECONDS.sleep(1);

                p.poll();
                System.out.println("Poll Data " + p.size());

                // 緩醒生產者
                pro.signal();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                r.unlock();
            }
        }
    }
}

這裡我們只說明不同點，首先 ^1. 使用鎖不再是透過 synchronized 關鍵字，而是使用 lock、unlock 操作，再來 ^2. 鎖的等待使用生產者自身的 Condition#await 方法

在生產完元素後使用消費者的 Condition#signal 喚醒消費者

class rProducer extends rCommonReentrant implements Runnable {

    @Override
    public void run() {
        while(true) {

            try {
                r.lock();

                while(p.size() == 10) {
                     System.out.println("Queue Full can not offer anymore");
                     pro.await();	
                }
                
                // 故意拖延生產
                TimeUnit.SECONDS.sleep(1);

                p.offer("Hello");
                System.out.println("offer Data " + p.size());

                // 喚醒消費者
                con.signal();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                r.unlock();
            }
        }
    }
}

public class reenBlocking {

    public static void main(String[] args) {
        rConsumer c = new rConsumer();
        rProducer p = new rProducer();

        new Thread(c).start();
        new Thread(p).start();
    }

}