Java 併發--- 阻塞隊列

對於多線程編程，java中有同步容器（HashTable，Vector），併發容器（ConcurrentHashMap、CopyOnWriteArrayList），還有阻塞隊列。

非阻塞隊列，好比PriorityQueue、LinkedList，一個很大問題就是：它不會對當前線程產生阻塞，那麼在面對相似消費者-生產者的模型時，就必須額外地實現同步策略以及線程間喚醒策略，這個實現起來就很是麻煩。可是有了阻塞隊列就不同了，它會對當前線程產生阻塞，好比一個線程從一個空的阻塞隊列中取元素，此時線程會被阻塞直到阻塞隊列中有了元素。當隊列中有元素後，被阻塞的線程會自動被喚醒（不須要咱們編寫代碼去喚醒）。這樣提供了極大的方便性。

幾種主要的阻塞隊列

ArrayBlockingQueue：基於數組實現的一個阻塞隊列，在建立ArrayBlockingQueue對象時必須制定容量大小。而且能夠指定公平性與非公平性，默認狀況下爲非公平的，即不保證等待時間最長的隊列最優先可以訪問隊列。

LinkedBlockingQueue：基於鏈表實現的一個阻塞隊列，在建立LinkedBlockingQueue對象時若是不指定容量大小，則默認大小爲Integer.MAX_VALUE。

PriorityBlockingQueue：以上2種隊列都是先進先出隊列，而PriorityBlockingQueue卻不是，它會按照元素的優先級對元素進行排序，按照優先級順序出隊，每次出隊的元素都是優先級最高的元素。注意，此阻塞隊列爲無界阻塞隊列，即容量沒有上限（經過源碼就能夠知道，它沒有容器滿的信號標誌），前面2種都是有界隊列。

DelayQueue：基於PriorityQueue，一種延時阻塞隊列，DelayQueue中的元素只有當其指定的延遲時間到了，纔可以從隊列中獲取到該元素。DelayQueue也是一個無界隊列，所以往隊列中插入數據的操做（生產者）永遠不會被阻塞，而只有獲取數據的操做（消費者）纔會被阻塞。

阻塞隊列中的方法 VS 非阻塞隊列中的方法

非阻塞隊列中的幾個主要方法

add(E e):將元素e插入到隊列末尾，若是插入成功，則返回true；若是插入失敗（即隊列已滿），則會拋出異常；

remove()：移除隊首元素，若移除成功，則返回true；若是移除失敗（隊列爲空），則會拋出異常；

offer(E e)：將元素e插入到隊列末尾，若是插入成功，則返回true；若是插入失敗（即隊列已滿），則返回false；

poll()：移除並獲取隊首元素，若成功，則返回隊首元素；不然返回null；

peek()：獲取隊首元素，若成功，則返回隊首元素；不然返回null

對於非阻塞隊列，通常狀況下建議使用offer、poll和peek三個方法，不建議使用add和remove方法。由於使用offer、poll和peek三個方法能夠經過返回值判斷操做成功與否，而使用add和remove方法卻不能達到這樣的效果。注意，非阻塞隊列中的方法都沒有進行同步措施。

阻塞隊列中的幾個主要方法

阻塞隊列包括了非阻塞隊列中的大部分方法，上面列舉的5個方法在阻塞隊列中都存在，可是要注意這5個方法在阻塞隊列中都進行了同步措施。除此以外，阻塞隊列提供了另外4個很是有用的方法：

put(E e) ：向隊尾存入元素，若是隊列滿，則等待；

take() ：從隊首取元素，若是隊列爲空，則等待；

offer(E e,long timeout, TimeUnit unit) ：向隊尾存入元素，若是隊列滿，則等待必定的時間，當時間期限達到時，若是尚未插入成功，則返回false；不然返回true；

poll(long timeout, TimeUnit unit) ：從隊首取元素，若是隊列空，則等待必定的時間，當時間期限達到時，若是取到，則返回null；不然返回取得的元素；

阻塞隊列的實現原理

// TODO

示例

使用Object.wait()和Object.notify()、非阻塞隊列實現生產者-消費者模式，wait()和notify()主要用來實現線程間通訊。代碼以下：

public class Test {
    private int queueSize = 10;
    private PriorityQueue<Integer> queue = new PriorityQueue<Integer>(queueSize);
     
    public static void main(String[] args)  {
        Test test = new Test();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();
         
        producer.start();
        consumer.start();
    }
     
    class Consumer extends Thread{
        @Override
        public void run() {
            consume();
        }
         
        private void consume() {
            while(true){
                synchronized (queue) {
                    while(queue.size() == 0){
                        try {
                            System.out.println("隊列空，等待數據");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }
                    queue.poll();          //每次移走隊首元素
                    queue.notify();
                    System.out.println("從隊列取走一個元素，隊列剩餘"+queue.size()+"個元素");
                }
            }
        }
    }
     
    class Producer extends Thread{
        @Override
        public void run() {
            produce();
        }
         
        private void produce() {
            while(true){
                synchronized (queue) {
                    while(queue.size() == queueSize){
                        try {
                            System.out.println("隊列滿，等待有空餘空間");
                            queue.wait();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                            queue.notify();
                        }
                    }
                    queue.offer(1);        //每次插入一個元素
                    queue.notify();
                    System.out.println("向隊列取中插入一個元素，隊列剩餘空間："+(queueSize-queue.size()));
                }
            }
        }
    }
}

使用阻塞隊列實現的生產者-消費者模式。代碼以下：

public class Test {
    private int queueSize = 10;
    private ArrayBlockingQueue<Integer> queue = new ArrayBlockingQueue<Integer>(queueSize);
     
    public static void main(String[] args)  {
        Test test = new Test();
        Producer producer = test.new Producer();
        Consumer consumer = test.new Consumer();
         
        producer.start();
        consumer.start();
    }
     
    class Consumer extends Thread{
        @Override
        public void run() {
            consume();
        }
         
        private void consume() {
            while(true){
                try {
                    queue.take();
                    System.out.println("從隊列取走一個元素，隊列剩餘"+queue.size()+"個元素");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
     
    class Producer extends Thread{
        @Override
        public void run() {
            produce();
        }
         
        private void produce() {
            while(true){
                try {
                    queue.put(1);
                    System.out.println("向隊列取中插入一個元素，隊列剩餘空間："+(queueSize-queue.size()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

使用阻塞隊列代碼要簡單得多，不須要再單獨考慮同步和線程間通訊的問題。

阻塞隊列使用最經典的場景就是socket客戶端數據的讀取和解析，讀取數據的線程不斷將數據放入隊列，而後解析線程不斷從隊列取數據解析。還有其餘相似的場景，只要符合生產者-消費者模型的均可以使用阻塞隊列。