本文主要講的是并發包中涉及到的集合,關于普通集合,請參考【java 集合概覽】
一、什么是blockingqueue
blockingqueue即阻塞隊列,從阻塞這個詞可以看出,在某些情況下對阻塞隊列的訪問可能會造成阻塞。被阻塞的情況主要有如下兩種:
1. 當隊列滿了的時候進行入隊列操作
2. 當隊列空了的時候進行出隊列操作
因此,當一個線程試圖對一個已經滿了的隊列進行入隊列操作時,它將會被阻塞,除非有另一個線程做了出隊列操作;同樣,當一個線程試圖對一個空隊列進行出隊列操作時,它將會被阻塞,除非有另一個線程進行了入隊列操作。
在java中,blockingqueue的接口位于java.util.concurrent
包中(在java5版本開始提供),由上面介紹的阻塞隊列的特性可知,阻塞隊列是線程安全的。
二、blockingqueue的用法
阻塞隊列主要用在生產者/消費者的場景,下面這幅圖展示了一個線程生產、一個線程消費的場景:
負責生產的線程不斷的制造新對象并插入到阻塞隊列中,直到達到這個隊列的上限值。隊列達到上限值之后生產線程將會被阻塞,直到消費的線程對這個隊列進行消費。同理,負責消費的線程不斷的從隊列中消費對象,直到這個隊列為空,當隊列為空時,消費線程將會被阻塞,除非隊列中有新的對象被插入。
三、blockingqueue接口中的方法
阻塞隊列一共有四套方法分別用來進行insert
、remove
和examine
,當每套方法對應的操作不能馬上執行時會有不同的反應,下面這個表格就分類列出了這些方法:
- | throws exception | special value | blocks | times out |
---|---|---|---|---|
insert | add(o) | offer(o) | put(o) | offer(o, timeout, timeunit) |
remove | remove(o) | poll() | take() | poll(timeout, timeunit) |
examine | element() | peek() |
這四套方法對應的特點分別是:
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的例子:
1
2
3
|
blockingqueue queue = new arrayblockingqueue( 1024 ); queue.put( "1" ); object object = queue.take(); |
4.2 delayqueue
delayqueue阻塞的是其內部元素,delayqueue中的元素必須實現 java.util.concurrent.delayed
接口,這個接口的定義非常簡單:
1
2
3
|
public interface delayed extends comparable<delayed> { long getdelay(timeunit unit); } |
getdelay()
方法的返回值就是隊列元素被釋放前的保持時間,如果返回0
或者一個負值
,就意味著該元素已經到期需要被釋放,此時delayedqueue會通過其take()
方法釋放此對象。
從上面delayed 接口定義可以看到,它還繼承了comparable
接口,這是因為delayedqueue中的元素需要進行排序,一般情況,我們都是按元素過期時間的優先級進行排序。
例1:為一個對象指定過期時間
首先,我們先定義一個元素,這個元素要實現delayed接口
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
|
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
1
2
3
4
5
6
7
8
9
|
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,誰先做完題目釋放誰
首先,我們構造一個學生對象
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
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 ; } } |
然后在構造一個教師對象對學生進行考試
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
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的例子:
1
2
3
4
|
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接口:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
|
public class priorityelement implements comparable<priorityelement> { private int priority; //定義優先級 priorityelement( int priority) { //初始化優先級 this .priority = priority; } @override public int compareto(priorityelement o) { //按照優先級大小進行排序 return priority >= o.getpriority() ? 1 : - 1 ; } public int getpriority() { return priority; } public void setpriority( int priority) { this .priority = priority; } @override public string tostring() { return "priorityelement [priority=" + priority + "]" ; } } |
然后我們把這些元素隨機設置優先級放入隊列中
1
2
3
4
5
6
7
8
9
10
11
12
13
|
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隊列內部僅允許容納一個元素。當一個線程插入一個元素后會被阻塞,除非這個元素被另一個線程消費。
以上就是本文的全部內容,希望對大家的學習有所幫助,也希望大家多多支持服務器之家。
原文鏈接:https://blog.csdn.net/suifeng3051/article/details/48807423