java多線程8：阻塞隊列與Fork/Join框架
隊列(Queue)，是一種數據結構。除了優先級隊列和LIFO隊列外，隊列都是以FIFO（先進先出）的方式對各個元素進行排序的。
BlockingQueue
而阻塞隊列BlockingQueue除了繼承隊列的所有方法外，還分別新增了支持阻塞的插入(put)和移除(take)方法。
下面我們分別看下JDK提供的七個阻塞隊列
ArrayBlockingQueue
ArrayBlockingQueue是一個用數組實現的有界阻塞隊列。此隊列按照先進先出（FIFO）的原則對元素進行排序。
看一下它的構造方法
/**
* Creates an {@code ArrayBlockingQueue} with the given (fixed)
* capacity and the specified access policy.
*
* @param capacity the capacity of this queue
* @param fair if {@code true} then queue accesses for threads blocked
* on insertion or removal, are processed in FIFO order;
* if {@code false} the access order is unspecified.
* @throws IllegalArgumentException if {@code capacity < 1}
*/
public ArrayBlockingQueue(int capacity, boolean fair) {
if (capacity <= 0)
throw new IllegalArgumentException();
this.items = new Object[capacity];
lock = new ReentrantLock(fair);
notEmpty = lock.newCondition();
notFull = lock.newCondition();
}
默認的構造方法是不保證線程公平的訪問隊列，所謂公平訪問隊列是指阻塞的線程，可以按照阻塞的先後順序訪問隊列，即先阻塞線程先訪問隊列。
非公平性是對先等待的線程是非公平的，當隊列可用時，阻塞的線程都可以爭奪訪問隊列的資格，有可能先阻塞的線程最後才訪問隊列。為了保證公平性，通常會降低吞吐量。
第15行 lock = new ReentrantLock(fair); 可以看出阻塞隊列的公平性也是通過ReentrantLock的公平鎖實現的，
第16行和第17行，我們又看到熟悉的 lock.newCondition()，在之前的一篇博客《java多線程6：ReentrantLock 》中，說到使用condition可以實現等待/通知模型，
也不難理解，如果想要實現支持阻塞的插入(put)和移除(take)方法，就可以通過condition實現線程間的通信。
LinkedBlockingQueue
LinkedBlockingQueue是一個用鏈表實現的有界阻塞隊列。此隊列的默認和最大長度為Integer.MAX_VALUE。此隊列按照先進先出的原則對元素進行排序。
offer(E e)： 如果隊列滿了，立即返回false，如果隊列沒滿，立即返回true
put(E e)：如果隊列滿了，一直阻塞，直到隊列不滿了或者線程被中斷
offer(E e, long timeout, TimeUnit unit)：在隊尾插入一個元素,如果隊列已滿，則進入等待，直到出現以下三種情況：被喚醒、等待時間超時、當前線程被中斷
poll()：如果沒有元素，直接返回null；如果有元素，出隊
take()：如果隊列有值，返回取出的值，如果隊列空了，一直阻塞，直到隊列不為空或者線程被中斷
poll(long timeout, TimeUnit unit)：如果隊列不空，出隊；如果隊列已空且已經超時，返回null；
如果隊列已空且時間未超時，則進入等待，直到出現以下三種情況：被喚醒、等待時間超時、當前線程被中斷
總結：LinkedBlockingQueue是允許兩個線程同時在兩端進行入隊或出隊的操作的，但一端同時只能有一個線程進行操作，這是通過兩把鎖來區分的；
為了維持底部數據的統一，引入了AtomicInteger的一個count變量，表示隊列中元素的個數。
count只能在兩個地方變化，一個是入隊的方法（可以+1），另一個是出隊的方法（可以-1），而AtomicInteger是原子安全的，所以也就確保了底層隊列的數據同步。
PriorityBlockingQueue
PriorityBlockingQueue是一個支持優先級的無界阻塞隊列。默認情況下元素採取自然順序升序排列。
也可以自定義類實現compareTo()方法來指定元素排序規則，或者初始化PriorityBlockingQueue時，指定構造參數Comparator來對元素進行排序。
需要注意的是不能保證同優先級元素的順序。
DelayQueue
DelayQueue是一個支持延時獲取元素的無界阻塞隊列。隊列使用PriorityQueue來實現。
隊列中的元素必須實現Delayed接口，在創建元素時可以指定多久才能從隊列中獲取當前元素。只有在延遲期滿時才能從隊列中提取元素。
使用場景：
訂單系統，對於30分鐘內未支付的訂單自動取消
緩存系統，可以用DelayQueue保存緩存元素的有效期，使用一個線程循環查詢DelayQueue，一旦能從DelayQueue中獲取元素時，表示緩存有效期到了
定時任務，使用DelayQueue保存當天將會執行的任務和執行時間，一旦從DelayQueue中獲取到任務就開始執行，比如TimerQueue就是使用DelayQueue實現的
SynchronousQueue
SynchronousQueue是一個不存儲元素的阻塞隊列。每一個put操作必須等待一個take操作，否則不能繼續添加元素。
/**
* Creates a {@code SynchronousQueue} with the specified fairness policy.
*
* @param fair if true, waiting threads contend in FIFO order for
* access; otherwise the order is unspecified.
*/
public SynchronousQueue(boolean fair) {
transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}
1、公平模式：TransferQueue
採用公平鎖，並配合一個FIFO隊列（Queue）來管理多餘的生產者和消費者
2、非公平模式：TransferStack
採用非公平鎖，並配合一個LIFO棧（Stack）來管理多餘的生產者和消費者，這也是SynchronousQueue默認的模式
LinkedTransferQueue
LinkedTransferQueue是一個由鏈表結構組成的無界阻塞TransferQueue隊列。相對於其他阻塞隊列，LinkedTransferQueue多了tryTransfer和transfer方法。
（1）transfer方法如果當前有消費者正在等待接收元素（消費者使用take()方法或帶時間限制的poll()方法時），transfer方法可以把生產者傳入的元素立刻transfer（傳輸）給消費者。
如果沒有消費者在等待接收元素，transfer方法會將元素存放在隊列的tail節點，並等到該元素被消費者消費了才返回。transfer方法的關鍵代碼如下。
（2）tryTransfer方法tryTransfer方法是用來試探生產者傳入的元素是否能直接傳給消費者。如果沒有消費者等待接收元素，則返回false。
和transfer方法的區別是tryTransfer方法無論消費者是否接收，方法立即返回，而transfer方法是必須等到消費者消費了才返回。
LinkedBlockingDeque
LinkedBlockingDeque是一個由鏈表結構組成的雙向阻塞隊列。所謂雙向隊列指的是可以從隊列的兩端插入和移出元素。
雙向隊列因為多了一個操作隊列的入口，在多線程同時入隊時，也就減少了一半的競爭。相比其他的阻塞隊列，
LinkedBlockingDeque多了addFirst、addLast、offerFirst、offerLast、peekFirst和peekLast等方法，
以First單詞結尾的方法，表示插入、獲取（peek）或移除雙端隊列的第一個元素。
以Last單詞結尾的方法，表示插入、獲取或移除雙端隊列的最後一個元素。
另外，插入方法add等同於addLast，移除方法remove等效於removeFirst。
但是take方法卻等同於takeFirst，不知道是不是JDK的bug，使用時還是用帶有First和Last後綴的方法更清楚。
在初始化LinkedBlockingDeque時可以設置容量防止其過度膨脹。另外，雙向阻塞隊列可以運用在「工作竊取」模式中。
Fork/Join框架
RecursiveAction：用於沒有返回結果的任務。
RecursiveTask：用於有返回結果的任務。
我們看一下如何用Fork/Join框架實現計算 1+2+3+…+100
public class CountTask extends RecursiveTask<Integer> {
// 閾值，設置每個子任務將執行多少計算任務
private static final int THRESHOLD = 2;
private int start;
private int end;
public CountTask(int start, int end) {
this.start = start;
this.end = end;
}
/**
* The main computation performed by this task.
*
* @return the result of the computation
*/
@Override
protected Integer compute() {
int sum = 0;
if (end - start > THRESHOLD) {
int mid = (start + end) / THRESHOLD;
CountTask leftTask = new CountTask(start, mid);
CountTask rightTask = new CountTask(mid + 1, end);
// 執行子任務
leftTask.fork();
rightTask.fork();
// 獲取子任務執行結果
Integer leftSum = leftTask.join();
Integer rightSum = rightTask.join();
sum = leftSum + rightSum;
} else {
for (int i = start; i <= end; i++) {
sum += i;
}
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool forkJoinPool = new ForkJoinPool();
CountTask countTask = new CountTask(1, 100);
ForkJoinTask<Integer> submit = forkJoinPool.submit(countTask);
try {
System.out.println(submit.get());
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
ForkJoinTask與一般任務的主要區別在於它需要實現compute方法，在這個方法里，首先需要判斷任務是否足夠小，如果足夠小就直接執行任務。
如果不足夠小，就必須分割成兩個子任務，每個子任務在調用fork方法時，又會進入compute方法，看看當前子任務是否需要繼續分割成子任務，
如果不需要繼續分割，則執行當前子任務並返回結果。使用join方法會等待子任務執行完並得到其結果。
參考文獻
1：《Java並發編程的藝術》
