Queue: 基本上,一個隊列就是一個先入先出(FIFO)的資料結構
Queue接口與List、Set同一級别,都是繼承了Collection接口。LinkedList實作了Deque接 口。
Queue的實作
1、沒有實作的阻塞接口的LinkedList: 實作了java.util.Queue接口和java.util.AbstractQueue接口
内置的不阻塞隊列: PriorityQueue 和 ConcurrentLinkedQueue
PriorityQueue 和 ConcurrentLinkedQueue 類在 Collection Framework 中加入兩個具體集合實作。
PriorityQueue 類實質上維護了一個有序清單。加入到 Queue 中的元素根據它們的天然排序(通過其 java.util.Comparable 實作)或者根據傳遞給構造函數的 java.util.Comparator 實作來定位。
ConcurrentLinkedQueue 是基于連結節點的、線程安全的隊列。并發通路不需要同步。因為它在隊列的尾部添加元素并從頭部删除它們,是以隻要不需要知道隊列的大 小, ConcurrentLinkedQueue 對公共集合的共享通路就可以工作得很好。收集關于隊列大小的資訊會很慢,需要周遊隊列。
2)實作阻塞接口的:
java.util.concurrent 中加入了 BlockingQueue 接口和五個阻塞隊列類。它實質上就是一種帶有一點扭曲的 FIFO 資料結構。不是立即從隊列中添加或者删除元素,線程執行操作阻塞,直到有空間或者元素可用。
五個隊列所提供的各有不同:
* ArrayBlockingQueue :一個由數組支援的有界隊列。
* LinkedBlockingQueue :一個由連結節點支援的可選有界隊列。
* PriorityBlockingQueue :一個由優先級堆支援的無界優先級隊列。
* DelayQueue :一個由優先級堆支援的、基于時間的排程隊列。
* SynchronousQueue :一個利用 BlockingQueue 接口的簡單聚集(rendezvous)機制。
下表顯示了jdk1.5中的阻塞隊列的操作:
add 增加一個元索 如果隊列已滿,則抛出一個IIIegaISlabEepeplian異常
remove 移除并傳回隊列頭部的元素 如果隊列為空,則抛出一個NoSuchElementException異常
element 傳回隊列頭部的元素 如果隊列為空,則抛出一個NoSuchElementException異常
offer 添加一個元素并傳回true 如果隊列已滿,則傳回false
poll 移除并返問隊列頭部的元素 如果隊列為空,則傳回null
peek 傳回隊列頭部的元素 如果隊列為空,則傳回null
put 添加一個元素 如果隊列滿,則阻塞
take 移除并傳回隊列頭部的元素 如果隊列為空,則阻塞
remove、element、offer 、poll、peek 其實是屬于Queue接口。
阻塞隊列的操作可以根據它們的響應方式分為以下三類:aad、removee和element操作在你試圖為一個已滿的隊列增加元素或從空隊列取得元素時 抛出異常。當然,在多線程程式中,隊列在任何時間都可能變成滿的或空的,是以你可能想使用offer、poll、peek方法。這些方法在無法完成任務時 隻是給出一個出錯示而不會抛出異常。
注意:poll和peek方法出錯進傳回null。是以,向隊列中插入null值是不合法的
最後,我們有阻塞操作put和take。put方法在隊列滿時阻塞,take方法在隊列空時阻塞。
LinkedBlockingQueue的容量是沒有上限的(說的不準确,在不指定時容量為Integer.MAX_VALUE,不要然的話在put時怎麼會受阻呢),但是也可以選擇指定其最大容量,它是基于連結清單的隊列,此隊列按 FIFO(先進先出)排序元素。
ArrayBlockingQueue在構造時需要指定容量, 并可以選擇是否需要公平性,如果公平參數被設定true,等待時間最長的線程會優先得到處理(其實就是通過将ReentrantLock設定為true來 達到這種公平性的:即等待時間最長的線程會先操作)。通常,公平性會使你在性能上付出代價,隻有在的确非常需要的時候再使用它。它是基于數組的阻塞循環隊 列,此隊列按 FIFO(先進先出)原則對元素進行排序。
PriorityBlockingQueue是一個帶優先級的 隊列,而不是先進先出隊列。元素按優先級順序被移除,該隊列也沒有上限(看了一下源碼,PriorityBlockingQueue是對 PriorityQueue的再次包裝,是基于堆資料結構的,而PriorityQueue是沒有容量限制的,與ArrayList一樣,是以在優先阻塞 隊列上put時是不會受阻的。雖然此隊列邏輯上是無界的,但是由于資源被耗盡,是以試圖執行添加操作可能會導緻 OutOfMemoryError),但是如果隊列為空,那麼取元素的操作take就會阻塞,是以它的檢索操作take是受阻的。另外,往入該隊列中的元 素要具有比較能力。
DelayQueue(基于PriorityQueue來實作的)是一個存放Delayed 元素的無界阻塞隊列,隻有在延遲期滿時才能從中提取元素。該隊列的頭部是延遲期滿後儲存時間最長的 Delayed 元素。如果延遲都還沒有期滿,則隊列沒有頭部,并且poll将傳回null。當一個元素的 getDelay(TimeUnit.NANOSECONDS) 方法傳回一個小于或等于零的值時,則出現期滿,poll就以移除這個元素了。此隊列不允許使用 null 元素。
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 {
/**
定義裝蘋果的籃子
*/
public static class Basket{
// 籃子,能夠容納3個蘋果
BlockingQueue<String> basket = new ArrayBlockingQueue<String>(3);
// 生産蘋果,放入籃子
public void produce() throws InterruptedException{
// put方法放入一個蘋果,若basket滿了,等到basket有位置
basket.put("An apple");
}
// 消費蘋果,從籃子中取走
public String consume() throws InterruptedException{
// get方法取出一個蘋果,若basket為空,等到basket有蘋果為止
String apple = basket.take();
return apple;
}
public int getAppleNumber(){
return basket.size();
}
}
// 測試方法
public static void testBasket() {
// 建立一個裝蘋果的籃子
final Basket basket = new Basket();
// 定義蘋果生産者
class Producer implements Runnable {
public void run() {
try {
while (true) {
// 生産蘋果
System.out.println("生産者準備生産蘋果:"
+ System.currentTimeMillis());
basket.produce();
System.out.println("生産者生産蘋果完畢:"
+ System.currentTimeMillis());
System.out.println("生産完後有蘋果:"+basket.getAppleNumber()+"個");
// 休眠300ms
Thread.sleep(300);
}
} catch (InterruptedException ex) {
}
}
}
// 定義蘋果消費者
class Consumer implements Runnable {
public void run() {
try {
while (true) {
// 消費蘋果
System.out.println("消費者準備消費蘋果:"
+ System.currentTimeMillis());
basket.consume();
System.out.println("消費者消費蘋果完畢:"
+ System.currentTimeMillis());
System.out.println("消費完後有蘋果:"+basket.getAppleNumber()+"個");
// 休眠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);
// 程式運作10s後,所有任務停止
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
}
service.shutdownNow();
}
public static void main(String[] args) {
BlockingQueueTest.testBasket();
}
}