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)	構造指定大小的有界隊列，指定為公平或非公平鎖，指定在初始化時加入一個集合

public ArrayBlockingQueue(int capacity) { 
　　this(capacity, false);//默認構造非公平鎖的阻塞隊列 
} 
public ArrayBlockingQueue(int capacity, boolean fair) { 
　　if (capacity <= 0)  
　　　　throw new IllegalArgumentException(); 
　　this.items = new Object[capacity]; 
　　lock = new ReentrantLock(fair);//初始化ReentrantLock重入鎖，出隊入隊擁有這同一個鎖 
　　notEmpty = lock.newCondition;//初始化非空等待隊列，有關Condition可參考《6.類似Object監視器方法的Condition接口》
　　notFull = lock.newCondition;//初始化非滿等待隊列 
} 
public ArrayBlockingQueue(int capacity, boolean fair, Collecation<? extends E> c) { 
　　this(capacity, fair); 
　　final ReentrantLock lock = this.lock; 
　　lock.lock();//注意在這個地方需要獲得鎖，這為什么需要獲取鎖的操作呢？ 
　　try { 
　　　　int i = 0; 
　　　　try { 
　　　　　　for (E e : c) { 
　　　　　　　　checkNotNull(e); 
　　　　　　　　item[i++] = e;//將集合添加進數組構成的隊列中 
　　　　　　} 
　　　　} catch (ArrayIndexOutOfBoundsException ex) { 
　　　　　　throw new IllegalArgumentException(); 
　　　　} 
　　　　count = i;//隊列中的實際數據數量 
　　　　putIndex = (i == capacity) ? 0 : i; 
　　} finally { 
　　　　lock.unlock(); 
　　} 
} 

　　在第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並沒有多大的用處。