java線程并發(fā)blockingqueue類(lèi)使用示例

更新時(shí)間：2014年01月16日 09:52:09 作者：

BlockingQueue是一種特殊的Queue，若BlockingQueue是空的，從BlockingQueue取東西的操作將會(huì)被阻斷進(jìn)入等待狀態(tài)直到BlocingkQueue進(jìn)了新貨才會(huì)被喚醒,下面是用BlockingQueue來(lái)實(shí)現(xiàn)Producer和Consumer的例子

如果BlockingQueue是滿(mǎn)的任何試圖往里存東西的操作也會(huì)被阻斷進(jìn)入等待狀態(tài)，直到BlockingQueue里有新的空間才會(huì)被喚醒繼續(xù)操作。

BlockingQueue提供的方法主要有：

add(anObject): 把a(bǔ)nObject加到BlockingQueue里，如果BlockingQueue可以容納返回true，否則拋出IllegalStateException異常。
offer(anObject)：把a(bǔ)nObject加到BlockingQueue里，如果BlockingQueue可以容納返回true，否則返回false。
put(anObject)：把a(bǔ)nObject加到BlockingQueue里，如果BlockingQueue沒(méi)有空間，調(diào)用此方法的線程被阻斷直到BlockingQueue里有新的空間再繼續(xù)。
poll(time)：取出BlockingQueue里排在首位的對(duì)象，若不能立即取出可等time參數(shù)規(guī)定的時(shí)間。取不到時(shí)返回null。
take()：取出BlockingQueue里排在首位的對(duì)象，若BlockingQueue為空，阻斷進(jìn)入等待狀態(tài)直到BlockingQueue有新的對(duì)象被加入為止。

根據(jù)不同的需要BlockingQueue有4種具體實(shí)現(xiàn)：

（1）ArrayBlockingQueue：規(guī)定大小的BlockingQueue，其構(gòu)造函數(shù)必須帶一個(gè)int參數(shù)來(lái)指明其大小。其所含的對(duì)象是以FIFO（先入先出）順序排序的。
（2）LinkedBlockingQueue：大小不定的BlockingQueue，若其構(gòu)造函數(shù)帶一個(gè)規(guī)定大小的參數(shù)，生成的BlockingQueue有大小限制，
若不帶大小參數(shù)，所生成的BlockingQueue的大小由Integer.MAX_VALUE來(lái)決定。其所含的對(duì)象是以FIFO（先入先出）順序排序的。
LinkedBlockingQueue和ArrayBlockingQueue比較起來(lái)，它們背后所用的數(shù)據(jù)結(jié)構(gòu)不一樣，
導(dǎo)致LinkedBlockingQueue的數(shù)據(jù)吞吐量要大于ArrayBlockingQueue，但在線程數(shù)量很大時(shí)其性能的可預(yù)見(jiàn)性低于ArrayBlockingQueue。
（3）PriorityBlockingQueue：類(lèi)似于LinkedBlockingQueue，但其所含對(duì)象的排序不是FIFO，而是依據(jù)對(duì)象的自然排序順序或者是構(gòu)造函數(shù)所帶的Comparator決定的順序。
（4）SynchronousQueue：特殊的BlockingQueue，對(duì)其的操作必須是放和取交替完成的。

復(fù)制代碼代碼如下:

package com.yao;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class BlockingQueueTest {

/**
定義裝蘋(píng)果的籃子
*/
public static class Basket{
  // 籃子，能夠容納3個(gè)蘋(píng)果
  BlockingQueue<String> basket = new ArrayBlockingQueue<String>(3);

  // 生產(chǎn)蘋(píng)果，放入籃子
  public void produce() throws InterruptedException{
   // put方法放入一個(gè)蘋(píng)果，若basket滿(mǎn)了，等到basket有位置
   basket.put("An apple");
  }
  // 消費(fèi)蘋(píng)果，從籃子中取走
  public String consume() throws InterruptedException{
   // get方法取出一個(gè)蘋(píng)果，若basket為空，等到basket有蘋(píng)果為止
   String apple = basket.take();
   return apple;
  }

  public int getAppleNumber(){
   return basket.size();
  }

}
//　測(cè)試方法
public static void testBasket() {
  // 建立一個(gè)裝蘋(píng)果的籃子
  final Basket basket = new Basket();
  // 定義蘋(píng)果生產(chǎn)者
  class Producer implements Runnable {
   public void run() {
    try {
     while (true) {
      // 生產(chǎn)蘋(píng)果
      System.out.println("生產(chǎn)者準(zhǔn)備生產(chǎn)蘋(píng)果："
        + System.currentTimeMillis());
      basket.produce();
      System.out.println("生產(chǎn)者生產(chǎn)蘋(píng)果完畢："
        + System.currentTimeMillis());
      System.out.println("生產(chǎn)完后有蘋(píng)果："+basket.getAppleNumber()+"個(gè)");
      // 休眠300ms
      Thread.sleep(300);
     }
    } catch (InterruptedException ex) {
    }
   }
  }
  // 定義蘋(píng)果消費(fèi)者
  class Consumer implements Runnable {
   public void run() {
    try {
     while (true) {
      // 消費(fèi)蘋(píng)果
      System.out.println("消費(fèi)者準(zhǔn)備消費(fèi)蘋(píng)果："
        + System.currentTimeMillis());
      basket.consume();
      System.out.println("消費(fèi)者消費(fèi)蘋(píng)果完畢："
        + System.currentTimeMillis());
      System.out.println("消費(fèi)完后有蘋(píng)果："+basket.getAppleNumber()+"個(gè)");
      // 休眠1000ms
      Thread.sleep(1000);
     }
    } catch (InterruptedException ex) {
    }
   }
  }

  ExecutorService service = Executors.newCachedThreadPool();
  Producer producer = new Producer();
  Consumer consumer = new Consumer();
  service.submit(producer);
  service.submit(consumer);
  // 程序運(yùn)行10s后，所有任務(wù)停止
  try {
   Thread.sleep(10000);
  } catch (InterruptedException e) {
  }
  service.shutdownNow();
}