BlockingQueue的使用

本例介紹一個特殊的隊列:BlockingQueue,如果BlockQueue是空的,從BlockingQueue取東西的操作將會被阻斷進入等待狀態,直到BlockingQueue進了東西才會被喚醒.同樣,如果BlockingQueue是滿的,任何試圖往里存東西的操作也會被阻斷進入等待狀態,直到BlockingQueue里有空間才會被喚醒繼續操作.
    使用BlockingQueue的關鍵技術點如下:
    1.BlockingQueue定義的常用方法如下:
        1)add(anObject):把anObject加到BlockingQueue里,即如果BlockingQueue可以容納,則返回true,否則報異常
        2)offer(anObject):表示如果可能的話,將anObject加到BlockingQueue里,即如果BlockingQueue可以容納,則返回true,否則返回false.
        3)put(anObject):把anObject加到BlockingQueue里,如果BlockQueue沒有空間,則調用此方法的線程被阻斷直到BlockingQueue里面有空間再繼續.
        4)poll(time):取走BlockingQueue里排在首位的對象,若不能立即取出,則可以等time參數規定的時間,取不到時返回null
        5)take():取走BlockingQueue里排在首位的對象,若BlockingQueue為空,阻斷進入等待狀態直到Blocking有新的對象被加入為止
    2.BlockingQueue有四個具體的實現類,根據不同需求,選擇不同的實現類
        1)ArrayBlockingQueue:規定大小的BlockingQueue,其構造函數必須帶一個int參數來指明其大小.其所含的對象是以FIFO(先入先出)順序排序的.
        2)LinkedBlockingQueue:大小不定的BlockingQueue,若其構造函數帶一個規定大小的參數,生成的BlockingQueue有大小限制,若不帶大小參數,所生成的BlockingQueue的大小由Integer.MAX_VALUE來決定.其所含的對象是以FIFO(先入先出)順序排序的
        3)PriorityBlockingQueue:類似於LinkedBlockQueue,但其所含對象的排序不是FIFO,而是依據對象的自然排序順序或者是構造函數的Comparator決定的順序.
        4)SynchronousQueue:特殊的BlockingQueue,對其的操作必須是放和取交替完成的.
    3.LinkedBlockingQueue和ArrayBlockingQueue比較起來,它們背后所用的數據結構不一樣,導致LinkedBlockingQueue的數據吞吐量要大於ArrayBlockingQueue,但在線程數量很大時其性能的可預見性低於ArrayBlockingQueue.      

　　下面是兩個使用BlockingQueue的例子：  

package com.thread;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueTest {
    public static void main(String[] args) {
        final BlockingQueue queue = new ArrayBlockingQueue(3);
        for(int i=0;i<2;i++){
            new Thread(){
                public void run(){
                    while(true){
                        try {
                            Thread.sleep((long)(Math.random()*1000));
                            System.out.println(Thread.currentThread().getName() + "准備放數據!");                            
                            queue.put(1);
                            System.out.println(Thread.currentThread().getName() + "已經放了數據，" +                             
                                        "隊列目前有" + queue.size() + "個數據");
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }

                    }
                }
                
            }.start();
        }
        
        new Thread(){
            public void run(){
                while(true){
                    try {
                        //將此處的睡眠時間分別改為100和1000，觀察運行結果
                        Thread.sleep(1000);
                        System.out.println(Thread.currentThread().getName() + "准備取數據!");
                        queue.take();
                        System.out.println(Thread.currentThread().getName() + "已經取走數據，" +                             
                                "隊列目前有" + queue.size() + "個數據");                    
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            
        }.start();            
    }
}

 

package com.thread;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class BlockingQueueCondition {

    public static void main(String[] args) {
        ExecutorService service = Executors.newSingleThreadExecutor();
        final Business3 business = new Business3();
        service.execute(new Runnable(){

            public void run() {
                for(int i=0;i<50;i++){
                    business.sub();
                }
            }
            
        });
        
        for(int i=0;i<50;i++){
            business.main();
        }
    }

}

class Business3{
    BlockingQueue subQueue = new ArrayBlockingQueue(1);
    BlockingQueue mainQueue = new ArrayBlockingQueue(1);
    //這里是匿名構造方法，只要new一個對象都會調用這個匿名構造方法，它與靜態塊不同，靜態塊只會執行一次，
    //在類第一次加載到JVM的時候執行
    //這里主要是讓main線程首先put一個，就有東西可以取，如果不加這個匿名構造方法put一個的話程序就死鎖了
    {
        try {
            mainQueue.put(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
    public void sub(){
        try
        {
            mainQueue.take();
            for(int i=0;i<10;i++){
                System.out.println(Thread.currentThread().getName() + " : " + i);
            }
            subQueue.put(1);
        }catch(Exception e){

        }
    }
    
    public void main(){
        
        try
        {
            subQueue.take();
            for(int i=0;i<5;i++){
                System.out.println(Thread.currentThread().getName() + " : " + i);
            }
            mainQueue.put(1);
        }catch(Exception e){
        }        
    }
}