10.併發包阻塞隊列之ArrayBlockingQueue

jdk1.7.0_79 

　　上一節中對併發包中的非阻塞隊列ConcurrentLinkedQueue的入隊、出隊作了一個簡要的分析，本文將對併發包中的阻塞隊列作一個簡要分析。 

　　Java併發包中的阻塞隊列一共7個，固然他們都是線程安全的。 

　　ArrayBlockingQueue：一個由數組結構組成的有界阻塞隊列。 

　　LinkedBlockingQueue：一個由鏈表結構組成的有界阻塞隊列。 

　　PriorityBlockingQueue：一個支持優先級排序的無界阻塞隊列。 

　　DealyQueue：一個使用優先級隊列實現的無界阻塞隊列。 

　　SynchronousQueue：一個不存儲元素的阻塞隊列。 

　　LinkedTransferQueue：一個由鏈表結構組成的無界阻塞隊列。 

　　LinkedBlockingDeque：一個由鏈表結構組成的雙向阻塞隊列。（摘自《Java併發編程的藝術》） 

　　在本文對ArrayBlockingQueue阻塞隊列作一個簡要解析 

　　對於ArrayLinkedQueue，放眼看過去其安全性的保證是由ReentrantLock保證的，有關ReentrantLock的解析可參考《5.Lock接口及其實現ReentrantLock》，在下文我也會適當的說起。 

　　首先來查看其構造函數： 

構造方法
public ArrayBlockingQueue(int capacity)	構造指定大小的有界隊列
public ArrayBlockingQueue(int capacity, boolean fair)	構造指定大小的有界隊列，指定爲公平或非公平鎖
public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c)	構造指定大小的有界隊列，指定爲公平或非公平鎖，指定在初始化時加入一個集合

 1 public ArrayBlockingQueue(int capacity) { 
 2 　　this(capacity, false);//默認構造非公平鎖的阻塞隊列 
 3 } 
 4 public ArrayBlockingQueue(int capacity, boolean fair) { 
 5 　　if (capacity <= 0)  
 6 　　　　throw new IllegalArgumentException(); 
 7 　　this.items = new Object[capacity]; 
 8 　　lock = new ReentrantLock(fair);//初始化ReentrantLock重入鎖，出隊入隊擁有這同一個鎖 
 9 　　notEmpty = lock.newCondition;//初始化非空等待隊列，有關Condition可參考《6.相似Object監視器方法的Condition接口》
10 　　notFull = lock.newCondition;//初始化非滿等待隊列 
11 } 
12 public ArrayBlockingQueue(int capacity, boolean fair, Collecation<? extends E> c) { 
13 　　this(capacity, fair); 
14 　　final ReentrantLock lock = this.lock; 
15 　　lock.lock();//注意在這個地方須要得到鎖，這爲何須要獲取鎖的操做呢？ 
16 　　try { 
17 　　　　int i = 0; 
18 　　　　try { 
19 　　　　　　for (E e : c) { 
20 　　　　　　　　checkNotNull(e); 
21 　　　　　　　　item[i++] = e;//將集合添加進數組構成的隊列中 
22 　　　　　　} 
23 　　　　} catch (ArrayIndexOutOfBoundsException ex) { 
24 　　　　　　throw new IllegalArgumentException(); 
25 　　　　} 
26 　　　　count = i;//隊列中的實際數據數量 
27 　　　　putIndex = (i == capacity) ? 0 : i; 
28 　　} finally { 
29 　　　　lock.unlock(); 
30 　　} 
31 } 

　　在第15行，源碼裏給了一句註釋： Lock only for visibility, not mutual exclusion。這句話的意思就是給出，這個鎖的操做並非爲了互斥操做，而是保證其可見性。線程T1是實例化ArrayBlockingQueue對象，T2是對實例化的ArrayBlockingQueue對象作入隊操做（固然要保證T1和T2的執行順序），若是不對它進行加鎖操做（加鎖會保證其可見性，也就是寫回主存），T1的集合c有可能只存在T1線程維護的緩存中，並無寫回主存，T2中實例化的ArrayBlockingQueue維護的緩存以及主存中並無集合c，此時就由於可見性形成數據不一致的狀況，引起線程安全問題。 

　　如下是ArrayBlockingQueue的一些出隊入隊操做。

隊列元素的插入

	拋出異常	返回值（非阻塞）	必定時間內返回值	返回值（阻塞）
插入	add(e)//隊列未滿時，返回true；隊列滿則拋出IllegalStateException(「Queue full」)異常——AbstractQueue	offer(e)//隊列未滿時，返回true；隊列滿時返回false。非阻塞當即返回。	offer(e, time, unit)//設定等待的時間，若是在指定時間內還不能往隊列中插入數據則返回false，插入成功返回true。	put(e)//隊列未滿時，直接插入沒有返回值；隊列滿時會阻塞等待，一直等到隊列未滿時再插入。

//ArrayBlockingQueue#add 
public boolean add(E e) { 
　　return super.add(e); 
} 

//AbstractQueue#add，這是一個模板方法，只定義add入隊算法骨架，成功時返回true，失敗時拋出IllegalStateException異常，具體offer實現交給子類實現。 
public boolean add(E e) { 
　　if (offer(e))//offer方法由Queue接口定義 
　　　　return true; 
　　else 
　　　　throw new IllegalStateException(); 
}

//ArrayBlockingQueue#offer，隊列未滿時返回true，滿時返回false 
public boolean offer(E e) { 
　　checkNotNull(e);//檢查入隊元素是否爲空 
　　final ReentrantLock lock = this.lock; 
　　lock.lock();//得到鎖，線程安全 
　　try { 
　　　　if (count == items.length)//隊列滿時，不阻塞等待，直接返回false 
　　　　　　return false; 
　　　　else { 
　　　　　　insert(e);//隊列未滿，直接插入 
　　　　　　return true; 
　　　　} 
　　} finally {
　　　　lock.unlock();
　　}
} 

//ArrayBlockingQueue#insert 
private void insert(E e) { 
　　items[putIndex] = x; 
　　putIndex = inc(putIndex); 
　　++count; 
　　notEmpty.signal();//喚醒非空等待隊列中的線程，有關Condition可參考《6.相似Object監視器方法的Condition接口》

 }

　　在這裏有幾個ArrayBlockingQueue成員變量。items即隊列的數組引用，putIndex表示等待插入的數組下標位置。當items[putIndex] = x將新元素插入隊列中後，調用inc將數組下標向後移動，若是隊列滿則將putIndex置爲0:

//ArrayBlockingQueue#inc 
private int inc(int i) { 
　　return (++i == items.length) ? 0 : i; 
} 

　　接着解析下put方法，阻塞插入隊列，當隊列滿時不會返回false，也不會拋出異常，而是一直阻塞等待，直到有空位可插入，但它可被中斷返回。

//ArrayBlockingQueue#put 
public void put(E e) throws InterruptedException { 
　　checkNotNull(e);//一樣檢查插入元素是否爲空 
　　final ReentrantLock lock = this.lock; 
　　lock.lockInterruptibly();//這裏並無調用lock方法，而是調用了可被中斷的lockInterruptibly，該方法可被線程中斷返回，lock不能被中斷返回。 
　　try { 
　　　　while (count == items.length) 
　　　　　　notFull.await();//當隊列滿時，使非滿等待隊列休眠 
　　　　insert(e);//此時表示隊列非滿，故插入元素，同時在該方法裏喚醒非空等待隊列 
　　} finally { 
　　　　lock.unlock(); 
　　} 
}  

隊列元素的刪除 

拋出異常	返回值（非阻塞）	必定時間內返回值	返回值（阻塞）
remove()//隊列不爲空時，返回隊首值並移除；隊列爲空時拋出NoSuchElementException()異常——AbstractQueue	poll()//隊列不爲空時返回隊首值並移除；隊列爲空時返回null。非阻塞當即返回。	poll(time, unit)//設定等待的時間，若是在指定時間內隊列還未孔則返回null，不爲空則返回隊首值	take(e)//隊列不爲空返回隊首值並移除；當隊列爲空時會阻塞等待，一直等到隊列不爲空時再返回隊首值。

//AbstractQueue#remove，這也是一個模板方法，定義刪除隊列元素的算法骨架，隊列中元素時返回具體元素，元素爲空時拋出異常，具體實現poll由子類實現， 
public E remove() { 
　　E x = poll();//poll方法由Queue接口定義 
　　if (x != null) 
　　　　return x; 
　　else 
　　　　throw new NoSuchElementException(); 
} 

//ArrayBlockingQueue#poll，隊列中有元素時返回元素，不爲空時返回null 
public E poll() { 
　　final ReentrantLock lock = this.lock; 
　　lock.lock(); 
　　try { 
　　　　return (count == 0) ? null : extract(); 
　　} finally { 
　　　　lock.unlock(); 
　　} 
}

//ArrayBlockingQueue#extract 
private E extract() { 
　　final Object[] items = this.items; 
　　E x = this.<E>cast(items[takeIndex]);//移除隊首元素 
　　items[takeIndex] = null;//將隊列數組中的第一個元素置爲null，便於GC回收 
　　takeIndex = inc(takeIndex); 
　　--count; 
　　notFull.signal();//喚醒非滿等待隊列線程 
　　return x; 
} 

　　對比add和offer方法，理解了上兩個方法後remove和poll實際不難理解，同理在理解了put阻塞插入隊列後，對比take阻塞刪除隊列元素一樣也很好理解。

//ArrayBlockQueue#take 
public E take() throws InterruptedException { 
　　final ReentrantLock lock = this.lock(); 
　　lock.lockInterrupted();//這裏並無調用lock方法，而是調用了可被中斷的lockInterruptibly，該方法可被線程中斷返回，lock不能被中斷返回。 
　　try { 
　　　　while (count == 0)//隊列元素爲空 
　　　　　　notEmpty.await();//非空等待隊列休眠 
　　　　return extract();//此時表示隊列非空，故刪除元素，同時在裏喚醒非滿等待隊列 
　　} finally { 
　　　　lock.unlock(); 
　　} 
} 

　　最後一個方法size。

public int size() { 
　　final ReentrantLock lock = this.lock; 
　　lock.lock(); 
　　try { 
　　　　return count; 
　　} finally { 
　　　　lock.unlock(); 
　　} 
}

　　能夠看到ArrayBlockingQueue隊列的size方法，是直接返回的count變量，它不像ConcurrentLinkedQueue，ConcurrentLinkedQueue的size則是每次會遍歷這個隊列，故ArrayBlockingQueue的size方法比ConcurrentLinkedQueue的size方法效率高。並且ConcurrentLinkedQueue的size方法並無加鎖！也就是說頗有可能其size並不許確，這在它的註釋中說明了ConcurrentLinkedQueue的size並無多大的用處。