Java並發之BlockingQueue的使用

本文主要講的是並發包中涉及到的集合，關於普通集合，請參考【java 集合概覽】

一、什麼是BlockingQueue

BlockingQueue即阻塞隊列，從阻塞這個詞可以看出，在某些情況下對阻塞隊列的訪問可能會造成阻塞。被阻塞的情況主要有如下兩種：

1. 當隊列滿了的時候進行入隊列操作
2. 當隊列空了的時候進行出隊列操作

因此，當一個線程試圖對一個已經滿了的隊列進行入隊列操作時，它將會被阻塞，除非有另一個線程做了出隊列操作；同樣，當一個線程試圖對一個空隊列進行出隊列操作時，它將會被阻塞，除非有另一個線程進行了入隊列操作。

在Java中，BlockingQueue的接口位於java.util.concurrent包中(在Java5版本開始提供)，由上面介紹的阻塞隊列的特性可知，阻塞隊列是線程安全的。

二、BlockingQueue的用法

阻塞隊列主要用在生產者/消費者的場景，下面這幅圖展示了一個線程生產、一個線程消費的場景：

負責生產的線程不斷的製造新對象並插入到阻塞隊列中，直到達到這個隊列的上限值。隊列達到上限值之後生產線程將會被阻塞，直到消費的線程對這個隊列進行消費。同理，負責消費的線程不斷的從隊列中消費對象，直到這個隊列為空，當隊列為空時，消費線程將會被阻塞，除非隊列中有新的對像被插入。

三、BlockingQueue接口中的方法

阻塞隊列一共有四套方法分別用來進行insert 、 remove和examine ，當每套方法對應的操作不能馬上執行時會有不同的反應，下面這個表格就分類列出了這些方法：

這四套方法對應的特點分別是：

1. ThrowsException：如果操作不能馬上進行，則拋出異常
2. SpecialValue：如果操作不能馬上進行，將會返回一個特殊的值，一般是true或者false
3. Blocks:如果操作不能馬上進行，操作會被阻塞
4. TimesOut:如果操作不能馬上進行，操作會被阻塞指定的時間，如果指定時間沒執行，則返回一個特殊值，一般是true或者false

需要注意的是，我們不能向BlockingQueue中插入null ，否則會報NullPointerException 。

四、BlockingQueue的實現類

BlockingQueue只是java.util.concurrent包中的一個接口，而在具體使用時，我們用到的是它的實現類，當然這些實現類也位於java.util.concurrent包中。在Java6中，BlockingQueue的實現類主要有以下幾種：

1. ArrayBlockingQueue
2. DelayQueue
3. LinkedBlockingQueue
4. PriorityBlockingQueue
5. SynchronousQueue

下面我們就分別介紹這幾個實現類。

4.1 ArrayBlockingQueue

ArrayBlockingQueue是一個有邊界的阻塞隊列，它的內部實現是一個數組。有邊界的意思是它的容量是有限的，我們必須在其初始化的時候指定它的容量大小，容量大小一旦指定就不可改變。

ArrayBlockingQueue是以先進先出的方式存儲數據，最新插入的對像是尾部，最新移出的對像是頭部。下面是一個初始化和使用ArrayBlockingQueue的例子：

 BlockingQueue queue = new ArrayBlockingQueue(1024);queue.put("1");Object object = queue.take();

4.2 DelayQueue

DelayQueue阻塞的是其內部元素，DelayQueue中的元素必須實現java.util.concurrent.Delayed接口，這個接口的定義非常簡單：

 public interface Delayed extends Comparable<Delayed> {long getDelay(TimeUnit unit);}

getDelay()方法的返回值就是隊列元素被釋放前的保持時間，如果返回0或者一個负值，就意味著該元素已經到期需要被釋放，此時DelayedQueue會通過其take()方法釋放此對象。

從上面Delayed 接口定義可以看到，它還繼承了Comparable接口，這是因為DelayedQueue中的元素需要進行排序，一般情況，我們都是按元素過期時間的優先級進行排序。

例1：為一個對象指定過期時間

首先，我們先定義一個元素，這個元素要實現Delayed接口

public class DelayedElement implements Delayed { private long expired; private long delay; private String name; DelayedElement(String elementName, long delay) { this. name = elementName; this. delay= delay; expired = ( delay + System. currentTimeMillis()); } @Override public int compareTo(Delayed o) { DelayedElement cached=(DelayedElement) o; return cached.getExpired()> expired?1:-1; } @Override public long getDelay(TimeUnit unit) { return ( expired - System. currentTimeMillis()); } @Override public String toString() { return "DelayedElement [delay=" + delay + ", name=" + name + "]"; } public long getExpired() { return expired; }}

設置這個元素的過期時間為3s

 public class DelayQueueExample { public static void main(String[] args) throws InterruptedException { DelayQueue<DelayedElement> queue= new DelayQueue<>(); DelayedElement ele= new DelayedElement( "cache 3 seconds",3000); queue.put( ele); System. out.println( queue.take()); }}

運行這個main函數，我們可以發現，我們需要等待3s之後才會打印這個對象。

其實DelayQueue應用場景很多，比如定時關閉連接、緩存對象，超時處理等各種場景，下面我們就拿學生考試為例讓大家更深入的理解DelayQueue的使用。

例2：把所有考試的學生看做是一個DelayQueue，誰先做完題目釋放誰

首先，我們構造一個學生對象

public class Student implements Runnable,Delayed{ private String name; //姓名private long costTime;//做試題的時間private long finishedTime;//完成時間public Student(String name, long costTime) { this. name = name; this. costTime= costTime; finishedTime = costTime + System. currentTimeMillis(); } @Override public void run() { System. out.println( name + " 交卷,用時" + costTime /1000); } @Override public long getDelay(TimeUnit unit) { return ( finishedTime - System. currentTimeMillis()); } @Override public int compareTo(Delayed o) { Student other = (Student) o; return costTime >= other. costTime?1:-1; }}

然後在構造一個教師對像對學生進行考試

public class Teacher { static final int STUDENT_SIZE = 30; public static void main(String[] args) throws InterruptedException { Random r = new Random(); //把所有學生看做一個延遲隊列DelayQueue<Student> students = new DelayQueue<Student>(); //構造一個線程池用來讓學生們“做作業” ExecutorService exec = Executors.newFixedThreadPool(STUDENT_SIZE); for ( int i = 0; i < STUDENT_SIZE; i++) { //初始化學生的姓名和做題時間students.put( new Student( "學生" + (i + 1), 3000 + r.nextInt(10000))); } //開始做題while(! students.isEmpty()){ exec.execute( students.take()); } exec.shutdown(); }}

我們看一下運行結果：

學生2 交卷,用時3
學生1 交卷,用時5
學生5 交卷,用時7
學生4 交卷,用時8
學生3 交卷,用時11

通過運行結果我們可以發現，每個學生在指定開始時間到達之後就會“交卷”（取決於getDelay()方法），並且是先做完的先交卷（取決於compareTo()方法）。

通過查看其源碼可以看到，DelayQueue內部實現用的是PriorityQueue和一個Lock：

4.3 LinkedBlockingQueue

LinkedBlockingQueue阻塞隊列大小的配置是可選的，如果我們初始化時指定一個大小，它就是有邊界的，如果不指定，它就是無邊界的。說是無邊界，其實是採用了默認大小為Integer.MAX_VALUE的容量。它的內部實現是一個鍊錶。

和ArrayBlockingQueue一樣，LinkedBlockingQueue 也是以先進先出的方式存儲數據，最新插入的對像是尾部，最新移出的對像是頭部。下面是一個初始化和使LinkedBlockingQueue的例子：

 BlockingQueue<String> unbounded = new LinkedBlockingQueue<String>();BlockingQueue<String> bounded = new LinkedBlockingQueue<String>(1024);bounded.put("Value");String value = bounded.take();

4.4 PriorityBlockingQueue

PriorityBlockingQueue是一個沒有邊界的隊列，它的排序規則和java.util.PriorityQueue一樣。需要注意，PriorityBlockingQueue中允許插入null對象。

所有插入PriorityBlockingQueue的對象必須實現java.lang.Comparable接口，隊列優先級的排序規則就是按照我們對這個接口的實現來定義的。

另外，我們可以從PriorityBlockingQueue獲得一個迭代器Iterator，但這個迭代器並不保證按照優先級順序進行迭代。

下面我們舉個例子來說明一下，首先我們定義一個對像類型，這個對象需要實現Comparable接口：

 public class PriorityElement implements Comparable<PriorityElement> {private int priority;//定義優先級PriorityElement(int priority) { //初始化優先級this.priority = priority;}@Overridepublic int compareTo(PriorityElement o) { //按照優先級大小進行排序return priority >= o.getPriority() ? 1 : -1;}public int getPriority() { return priority;}public void setPriority(int priority) { this.priority = priority;}@Overridepublic String toString() { return "PriorityElement [priority=" + priority + "]";}}

然後我們把這些元素隨機設置優先級放入隊列中

public class PriorityBlockingQueueExample {public static void main(String[] args) throws InterruptedException { PriorityBlockingQueue<PriorityElement> queue = new PriorityBlockingQueue<>(); for (int i = 0; i < 5; i++) { Random random=new Random(); PriorityElement ele = new PriorityElement(random.nextInt(10)); queue.put(ele); } while(!queue.isEmpty()){ System.out.println(queue.take()); }}}

看一下運行結果：

PriorityElement [priority=3]
PriorityElement [priority=4]
PriorityElement [priority=5]
PriorityElement [priority=8]
PriorityElement [priority=9]

4.5 SynchronousQueue

SynchronousQueue隊列內部僅允許容納一個元素。當一個線程插入一個元素後會被阻塞，除非這個元素被另一個線程消費。

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。