java多線程學習-java.util.concurrent詳解(四) BlockingQueue

7.BlockingQueue 

    “支援兩個附加操作的 Queue，這兩個操作是：擷取元素時等待隊列變為非空，以及存儲元素時等待空間變得可用。“ 

    這裡我們主要讨論BlockingQueue的最典型實作：LinkedBlockingQueue 和ArrayBlockingQueue。兩者的不同是底層的資料結構不夠，一個是連結清單，另外一個是數組。 

    後面将要單獨解釋其他類型的BlockingQueue和SynchronousQueue 

     BlockingQueue的經典用途是 生産者-消費者模式  

    代碼如下： 

Java代碼  

1. import java.util.Random;  
2. import java.util.concurrent.BlockingQueue;  
3. import java.util.concurrent.LinkedBlockingQueue;  
4. public class TestBlockingQueue {  
5.     public static void main(String[] args) {  
6.         final BlockingQueue<Integer> queue=new LinkedBlockingQueue<Integer>(3);  
7.         final Random random=new Random();  
8.         class Producer implements Runnable{  
9.             @Override  
10.             public void run() {  
11.                 while(true){  
12.                     try {  
13.                         int i=random.nextInt(100);  
14.                         queue.put(i);//當隊列達到容量時候，會自動阻塞的  
15.                         if(queue.size()==3)  
16.                         {  
17.                             System.out.println("full");  
18.                         }  
19.                     } catch (InterruptedException e) {  
20.                         e.printStackTrace();  
21.                     }  
22.                 }  
23.             }  
24.         }  
25.         class Consumer implements Runnable{  
26.             @Override  
27.             public void run() {  
28.                 while(true){  
29.                     try {  
30.                         queue.take();//當隊列為空時，也會自動阻塞  
31.                         Thread.sleep(1000);  
32.                     } catch (InterruptedException e) {  
33.                         e.printStackTrace();  
34.                     }  
35.                 }  
36.             }  
37.         }  
38.         new Thread(new Producer()).start();  
39.         new Thread(new Consumer()).start();  
40.     }  
41. }  

    總結：BlockingQueue使用時候特别注意take 和 put 

8. DelayQueue 

我們先來學習一下JDK1.5 API中關于這個類的詳細介紹： 

    “它是包含Delayed 元素的一個無界阻塞隊列，隻有在延遲期滿時才能從中提取元素。該隊列的頭部 是延遲期滿後儲存時間最長的 Delayed 元素。如果延遲都還沒有期滿，則隊列沒有頭部，并且 poll 将傳回 null。當一個元素的 getDelay(TimeUnit.NANOSECONDS) 方法傳回一個小于等于 0 的值時，将發生到期。即使無法使用 take 或 poll 移除未到期的元素，也不會将這些元素作為正常元素對待。例如，size 方法同時傳回到期和未到期元素的計數。此隊列不允許使用 null 元素。” 

    在現實生活中，很多DelayQueue的例子。就拿上海的SB會來說明，很多國家地區的開館時間不同。你很早就來到園區，然後急急忙忙地跑到一些心儀的館區，發現有些還沒開，你吃了閉門羹。 

    仔細研究DelayQueue，你會發現它其實就是一個PriorityQueue的封裝（按照delay時間排序），裡面的元素都實作了Delayed接口，相關操作需要判斷延時時間是否到了。 

    在實際應用中，有人拿它來管理跟實際相關的緩存、session等 

    下面我就通過 “上海SB會的例子來闡述DelayQueue的用法”  

代碼如下： 

Java代碼  

1. import java.util.Random;  
2. import java.util.concurrent.DelayQueue;  
3. import java.util.concurrent.Delayed;  
4. import java.util.concurrent.TimeUnit;  
5. public class TestDelayQueue {  
6.     private class Stadium implements Delayed  
7.     {  
8.         long trigger;  
9.         public Stadium(long i){  
10.             trigger=System.currentTimeMillis()+i;  
11.         }  
12.         @Override  
13.         public long getDelay(TimeUnit arg0) {  
14.             long n=trigger-System.currentTimeMillis();  
15.             return n;  
16.         }  
17.         @Override  
18.         public int compareTo(Delayed arg0) {  
19.             return (int)(this.getDelay(TimeUnit.MILLISECONDS)-arg0.getDelay(TimeUnit.MILLISECONDS));  
20.         }  
21.         public long getTriggerTime(){  
22.             return trigger;  
23.         }  
24.     }  
25.     public static void main(String[] args)throws Exception {  
26.         Random random=new Random();  
27.         DelayQueue<Stadium> queue=new DelayQueue<Stadium>();  
28.         TestDelayQueue t=new TestDelayQueue();  
29.         for(int i=0;i<5;i++){  
30.             queue.add(t.new Stadium(random.nextInt(30000)));  
31.         }  
32.         Thread.sleep(2000);  
33.         while(true){  
34.             Stadium s=queue.take();//延時時間未到就一直等待  
35.             if(s!=null){  
36.                 System.out.println(System.currentTimeMillis()-s.getTriggerTime());//基本上是等于0  
37.             }  
38.             if(queue.size()==0)  
39.                 break;  
40.         }  
41.     }  
42. }  

    總結：适用于需要延時操作的隊列管理 

9. SynchronousQueue 

    我們先來學習一下JDK1.5 API中關于這個類的詳細介紹： 

    “一種阻塞隊列，其中每個插入操作必須等待另一個線程的對應移除操作 ，反之亦然。同步隊列沒有任何内部容量，甚至連一個隊列的容量都沒有。不能在同步隊列上進行 peek，因為僅在試圖要移除元素時，該元素才存在；除非另一個線程試圖移除某個元素，否則也不能（使用任何方法）插入元素；也不能疊代隊列，因為其中沒 有元素可用于疊代。隊列的頭 是嘗試添加到隊列中的首個已排隊插入線程的元素；如果沒有這樣的已排隊線程，則沒有可用于移除的元素并且 poll() 将會傳回 null。對于其他 Collection 方法（例如 contains），SynchronousQueue 作為一個空 collection。此隊列不允許 null 元素。 

    同步隊列類似于 CSP 和 Ada 中使用的 rendezvous 信道。它非常适合于傳遞性設計，在這種設計中，在一個線程中運作的對象要将某些資訊、事件或任務傳遞給在另一個線程中運作的對象，它就必須與該對象同步。 “ 

    看起來很有意思吧。隊列竟然是沒有内部容量的。這個隊列其實是BlockingQueue的一種實作。每個插入操作必須等待另一個線程的對應移除操作，反之亦然。它給我們提供了線上程之間交換單一進制素的極輕量級方法 

    應用舉例：我們要在多個線程中傳遞一個變量。  

   代碼如下（其實就是生産者消費者模式） 

Java代碼  

1. import java.util.Arrays;  
2. import java.util.List;  
3. import java.util.concurrent.BlockingQueue;  
4. import java.util.concurrent.SynchronousQueue;  
5. public class TestSynchronousQueue {  
6.     class Producer implements Runnable {  
7.         private BlockingQueue<String> queue;  
8.         List<String> objects = Arrays.asList("one", "two", "three");  
9.         public Producer(BlockingQueue<String> q) {  
10.             this.queue = q;  
11.         }  
12.         @Override  
13.         public void run() {  
14.             try {  
15.                 for (String s : objects) {  
16.                     queue.put(s);// 産生資料放入隊列中  
17.                     System.out.printf("put:%s%n",s);  
18.                 }  
19.                 queue.put("Done");// 已完成的标志  
20.             } catch (InterruptedException e) {  
21.                 e.printStackTrace();  
22.             }  
23.         }  
24.     }  
25.     class Consumer implements Runnable {  
26.         private BlockingQueue<String> queue;  
27.         public Consumer(BlockingQueue<String> q) {  
28.             this.queue = q;  
29.         }  
30.         @Override  
31.         public void run() {  
32.             String obj = null;  
33.             try {  
34.                 while (!((obj = queue.take()).equals("Done"))) {  
35.                     System.out.println(obj);//從隊列中讀取對象  
36.                     Thread.sleep(3000);     //故意sleep，證明Producer是put不進去的  
37.                 }  
38.             } catch (InterruptedException e) {  
39.                 e.printStackTrace();  
40.             }  
41.         }  
42.     }  
43.     public static void main(String[] args) {  
44.         BlockingQueue<String> q=new SynchronousQueue<String>();  
45.         TestSynchronousQueue t=new TestSynchronousQueue();  
46.         new Thread(t.new Producer(q)).start();  
47.         new Thread(t.new Consumer(q)).start();  
48.     }  
49. }  

以下引用自：http://www.cnblogs.com/jackyuj/archive/2010/11/24/1886553.html

其中有很形象的解釋和示例

• 前言：

     在新增的Concurrent包中，BlockingQueue很好的解決了多線程中，如何高效安全“傳輸”資料的問題。通過這些高效并且線程安全的隊列 類，為我們快速搭建高品質的多線程程式帶來極大的便利。本文詳細介紹了BlockingQueue家庭中的所有成員，包括他們各自的功能以及常見使用場 景。

• 認識BlockingQueue

  阻塞隊列，顧名思義，首先它是一個隊列，而一個隊列在資料結構中所起的作用大緻如下圖所示：

  

  從上圖我們可以很清楚看到，通過一個共享的隊列，可以使得資料由隊列的一端輸入，從另外一端輸出；

  常用的隊列主要有以下兩種：（當然通過不同的實作方式，還可以延伸出很多不同類型的隊列，DelayQueue就是其中的一種）

  　　先進先出（FIFO）：先插入的隊列的元素也最先出隊列，類似于排隊的功能。從某種程度上來說這種隊列也展現了一種公平性。

  　　後進先出（LIFO）：後插入隊列的元素最先出隊列，這種隊列優先處理最近發生的事件。

        多線程環境中，通過隊列可以很容易實作資料共享，比如經典的“生産者”和“消費者”模型中，通過隊列可以很便利地實作兩者之間的資料共享。假設我們有若 幹生産者線程，另外又有若幹個消費者線程。如果生産者線程需要把準備好的資料共享給消費者線程，利用隊列的方式來傳遞資料，就可以很友善地解決他們之間的 資料共享問題。但如果生産者和消費者在某個時間段内，萬一發生資料處理速度不比對的情況呢？理想情況下，如果生産者産出資料的速度大于消費者消費的速度， 并且當生産出來的資料累積到一定程度的時候，那麼生産者必須暫停等待一下（阻塞生産者線程），以便等待消費者線程把累積的資料處理完畢，反之亦然。然而， 在concurrent包釋出以前，在多線程環境下，我們每個程式員都必須去自己控制這些細節，尤其還要兼顧效率和線程安全，而這會給我們的程式帶來不小 的複雜度。好在此時，強大的concurrent包橫空出世了，而他也給我們帶來了強大的BlockingQueue。（在多線程領域：所謂阻塞，在某些 情況下會挂起線程（即阻塞），一旦條件滿足，被挂起的線程又會自動被喚醒）

  下面兩幅圖示範了BlockingQueue的兩個常見阻塞場景：

  

  　　　　　　　如上圖所示：當隊列中沒有資料的情況下，消費者端的所有線程都會被自動阻塞（挂起），直到有資料放入隊列。

  

  　　　如上圖所示：當隊列中填滿資料的情況下，生産者端的所有線程都會被自動阻塞（挂起），直到隊列中有空的位置，線程被自動喚醒。

       這也是我們在多線程環境下，為什麼需要BlockingQueue的原因。作為BlockingQueue的使用者，我們再也不需要關心什麼時候需要阻塞 線程，什麼時候需要喚醒線程，因為這一切BlockingQueue都給你一手包辦了。既然BlockingQueue如此神通廣大，讓我們一起來見識下 它的常用方法：

  BlockingQueue的核心方法：

  放入資料：

  　　offer(anObject):表示如果可能的話,将anObject加到BlockingQueue裡,即如果BlockingQueue可以容納,

  　　　　則傳回true,否則傳回false.（本方法不阻塞目前執行方法的線程）

  　　offer(E o, long timeout, TimeUnit unit),可以設定等待的時間，如果在指定的時間内，還不能往隊列中

  　　　　加入BlockingQueue，則傳回失敗。

  　　put(anObject):把anObject加到BlockingQueue裡,如果BlockQueue沒有空間,則調用此方法的線程被阻斷

  　　　　直到BlockingQueue裡面有空間再繼續.

  擷取資料：

  　　poll(time):取走BlockingQueue裡排在首位的對象,若不能立即取出,則可以等time參數規定的時間,

  　　　　取不到時傳回null;

  　　poll(long timeout, TimeUnit unit)：從BlockingQueue取出一個隊首的對象，如果在指定時間内，

  　　　　隊列一旦有資料可取，則立即傳回隊列中的資料。否則知道時間逾時還沒有資料可取，傳回失敗。

  　　take():取走BlockingQueue裡排在首位的對象,若BlockingQueue為空,阻斷進入等待狀态直到

  　　　　BlockingQueue有新的資料被加入; 

  　　drainTo():一次性從BlockingQueue擷取所有可用的資料對象（還可以指定擷取資料的個數）， 

  　　　　通過該方法，可以提升擷取資料效率；不需要多次分批加鎖或釋放鎖。

• 常見BlockingQueue

  在了解了BlockingQueue的基本功能後，讓我們來看看BlockingQueue家庭大緻有哪些成員？ 

  

• BlockingQueue成員詳細介紹

  1. ArrayBlockingQueue

        基于數組的阻塞隊列實作，在ArrayBlockingQueue内部，維護了一個定長數組，以便緩存隊列中的資料對象，這是一個常用的阻塞隊列，除了 一個定長數組外，ArrayBlockingQueue内部還儲存着兩個整形變量，分别辨別着隊列的頭部和尾部在數組中的位置。

  　　 ArrayBlockingQueue在生産者放入資料和消費者擷取資料，都是共用同一個鎖對象，由此也意味着兩者無法真正并行運作，這點尤其不同于 LinkedBlockingQueue；按照實作原理來分析，ArrayBlockingQueue完全可以采用分離鎖，進而實作生産者和消費者操作的 完全并行運作。Doug Lea之是以沒這樣去做，也許是因為ArrayBlockingQueue的資料寫入和擷取操作已經足夠輕巧，以至于引入獨立的鎖機制，除了給代碼帶來額 外的複雜性外，其在性能上完全占不到任何便宜。 ArrayBlockingQueue和LinkedBlockingQueue間還有一個明顯的不同之處在于，前者在插入或删除元素時不會産生或銷毀任 何額外的對象執行個體，而後者則會生成一個額外的Node對象。這在長時間内需要高效并發地處理大批量資料的系統中，其對于GC的影響還是存在一定的差別。而 在建立ArrayBlockingQueue時，我們還可以控制對象的内部鎖是否采用公平鎖，預設采用非公平鎖。

  2. LinkedBlockingQueue

        基于連結清單的阻塞隊列，同ArrayListBlockingQueue類似，其内部也維持着一個資料緩沖隊列（該隊列由一個連結清單構成），當生産者往隊列 中放入一個資料時，隊列會從生産者手中擷取資料，并緩存在隊列内部，而生産者立即傳回；隻有當隊列緩沖區達到最大值緩存容量時 （LinkedBlockingQueue可以通過構造函數指定該值），才會阻塞生産者隊列，直到消費者從隊列中消費掉一份資料，生産者線程會被喚醒，反 之對于消費者這端的處理也基于同樣的原理。而LinkedBlockingQueue之是以能夠高效的處理并發資料，還因為其對于生産者端和消費者端分别 采用了獨立的鎖來控制資料同步，這也意味着在高并發的情況下生産者和消費者可以并行地操作隊列中的資料，以此來提高整個隊列的并發性能。

  作為開 發 者，我們需要注意的是，如果構造一個LinkedBlockingQueue對象，而沒有指定其容量大小，LinkedBlockingQueue會預設 一個類似無限大小的容量（Integer.MAX_VALUE），這樣的話，如果生産者的速度一旦大于消費者的速度，也許還沒有等到隊列滿阻塞産生，系統 記憶體就有可能已被消耗殆盡了。

  ArrayBlockingQueue和LinkedBlockingQueue是兩個最普通也是最常用的阻塞隊列，一般情況下，在處理多線程間的生産者消費者問題，使用這兩個類足以。

  下面的代碼示範了如何使用BlockingQueue：

  01	import java.util.concurrent.BlockingQueue;

  02	import java.util.concurrent.ExecutorService;

  03	import java.util.concurrent.Executors;

  04	import java.util.concurrent.LinkedBlockingQueue;

  05

  06

  09	public class BlockingQueueTest {

  10

  11	public static void main(String[] args)  throws InterruptedException {

  12	// 聲明一個容量為10的緩存隊列

  13	BlockingQueue<String> queue =  new LinkedBlockingQueue<String>( 10 );

  14

  15	Producer producer1 =  new Producer(queue);

  16	Producer producer2 =  new Producer(queue);

  17	Producer producer3 =  new Producer(queue);

  18	Consumer consumer =  new Consumer(queue);

  19

  20	// 借助Executors

  21	ExecutorService service = Executors.newCachedThreadPool();

  22

  // 啟動線程

  23	service.execute(producer1);

  24	service.execute(producer2);

  25	service.execute(producer3);

  26	service.execute(consumer);

  27

  28

  // 執行10s

  29	Thread.sleep( 10 *  1000 );

  30	producer1.stop();

  31	producer2.stop();

  32	producer3.stop();

  33

  34	Thread.sleep( 2000 );

  35	// 退出Executor

  36	service.shutdown();

  37

  }

  38

  }

  01	import java.util.Random;

  02	import java.util.concurrent.BlockingQueue;

  03	import java.util.concurrent.TimeUnit;

  04

  05

  10	public class Consumer  implements Runnable {

  11

  12	public Consumer(BlockingQueue<String> queue) {

  13	this .queue = queue;

  14

  }

  15

  16	public void run() {

  17	System.out.println( "啟動消費者線程！" );

  18	Random r =  new Random();

  19	boolean isRunning =  true ;

  20

  try {

  21	while (isRunning) {

  22	System.out.println( "正從隊列擷取資料..." );

  23	String data = queue.poll( 2 , TimeUnit.SECONDS);

  24	if ( null != data) {

  25	System.out.println( "拿到資料：" + data);

  26	System.out.println( "正在消費資料：" + data);

  27	Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));

  28

  }  else {

  29	// 超過2s還沒資料，認為所有生産線程都已經退出，自動退出消費線程。

  30	isRunning =  false ;

  31

  }

  32

  }

  33	}  catch (InterruptedException e) {

  34	e.printStackTrace();

  35	Thread.currentThread().interrupt();

  36	}  finally {

  37	System.out.println( "退出消費者線程！" );

  38

  }

  39

  }

  40

  41	private BlockingQueue<String> queue;

  42	private static final int DEFAULT_RANGE_FOR_SLEEP =  1000 ;

  43

  }

  44

  45	import java.util.Random;

  46	import java.util.concurrent.BlockingQueue;

  47	import java.util.concurrent.TimeUnit;

  48	import java.util.concurrent.atomic.AtomicInteger;

  49

  50

  55	public class Producer  implements Runnable {

  56

  57	public Producer(BlockingQueue queue) {

  58	this .queue = queue;

  59

  }

  60

  61	public void run() {

  62	String data =  null ;

  63	Random r =  new Random();

  64

  65	System.out.println( "啟動生産者線程！" );

  66

  try {

  67	while (isRunning) {

  68	System.out.println( "正在生産資料..." );

  69	Thread.sleep(r.nextInt(DEFAULT_RANGE_FOR_SLEEP));

  70

  71	data =  "data:" + count.incrementAndGet();

  72	System.out.println( "将資料：" + data +  "放入隊列..." );

  73	if (!queue.offer(data,  2 , TimeUnit.SECONDS)) {

  74	System.out.println( "放入資料失敗：" + data);

  75

  }

  76

  }

  77	}  catch (InterruptedException e) {

  78	e.printStackTrace();

  79	Thread.currentThread().interrupt();

  80	}  finally {

  81	System.out.println( "退出生産者線程！" );

  82

  }

  83

  }

  84

  85	public void stop() {

  86	isRunning =  false ;

  87

  }

  88

  89	private volatile boolean isRunning               =  true ;

  90	private BlockingQueue queue;

  91	private static AtomicInteger  count                   =  new AtomicInteger();

  92	private static final int DEFAULT_RANGE_FOR_SLEEP =  1000 ;

  93

  94

  }

• 3. DelayQueue

        DelayQueue中的元素隻有當其指定的延遲時間到了，才能夠從隊列中擷取到該元素。DelayQueue是一個沒有大小限制的隊列，是以往隊列中插入資料的操作（生産者）永遠不會被阻塞，而隻有擷取資料的操作（消費者）才會被阻塞。

  使用場景：

  　　DelayQueue使用場景較少，但都相當巧妙，常見的例子比如使用一個DelayQueue來管理一個逾時未響應的連接配接隊列。

  4. PriorityBlockingQueue

        基于優先級的阻塞隊列（優先級的判斷通過構造函數傳入的Compator對象來決定），但需要注意的是PriorityBlockingQueue并不 會阻塞資料生産者，而隻會在沒有可消費的資料時，阻塞資料的消費者。是以使用的時候要特别注意，生産者生産資料的速度絕對不能快于消費者消費資料的速度， 否則時間一長，會最終耗盡所有的可用堆記憶體空間。在實作PriorityBlockingQueue時，内部控制線程同步的鎖采用的是公平鎖。

  5. SynchronousQueue

        一種無緩沖的等待隊列，類似于無中介的直接交易，有點像原始社會中的生産者和消費者，生産者拿着産品去集市銷售給産品的最終消費者，而消費者必須親自去 集市找到所要商品的直接生産者，如果一方沒有找到合适的目标，那麼對不起，大家都在集市等待。相對于有緩沖的BlockingQueue來說，少了一個中 間經銷商的環節（緩沖區），如果有經銷商，生産者直接把産品批發給經銷商，而無需在意經銷商最終會将這些産品賣給那些消費者，由于經銷商可以庫存一部分商 品，是以相對于直接交易模式，總體來說采用中間經銷商的模式會吞吐量高一些（可以批量買賣）；但另一方面，又因為經銷商的引入，使得産品從生産者到消費 中間增加了額外的交易環節，單個産品的及時響應性能可能會降低。

  　　聲明一個SynchronousQueue有兩種不同的方式，它們之間有着不太一樣的行為。公平模式和非公平模式的差別:

  　　如果采用公平模式：SynchronousQueue會采用公平鎖，并配合一個FIFO隊列來阻塞多餘的生産者和消費者，進而體系整體的公平政策；

  　 　但如果是非公平模式（SynchronousQueue預設）：SynchronousQueue采用非公平鎖，同時配合一個LIFO隊列來管理多餘的 生産者和消費者，而後一種模式，如果生産者和消費者的處理速度有差距，則很容易出現饑渴的情況，即可能有某些生産者或者是消費者的資料永遠都得不到處理。

• 小結

  　　BlockingQueue不光實作了一個完整隊列所具有的基本功能，同時在多線程環境下，他還自動管理了多線間的自動等待于喚醒功能，進而使得程式員可以忽略這些細節，關注更進階的功能。