併發編程之線程池及隊列
問題：
- 線程池作用，主要實現類，並說出實現類場景以及區別。
- ThreadPoolExecutor使用場景。以及原理。
- Executor拒絕策略說的是什麼？
- 無界阻塞延遲隊列delayqueue原理是什麼？
- CyclicBarrier和CountDownLatch的區別？
線程池作用，主要實現類，並說出實現類場景以及區別
作用
- 減少在創建和銷燬線程上所花的時間以及系統資源的開銷。
- 如果不使用線程池，可能造成系統創建大量線程。
線程池種類
newCachedThreadPool
用newCachedThreadPool()方法創建該線程池對象，創建之初裡面一個線程都沒有，當execute方法或submit方法向線程池提交任務時，會自動新建線程；如果線程池中有空餘線程，則不會新建；這種線程池一般最多情況可以容納幾萬個線程，裡面的線程空餘60s會被回收。
適用場景：執行很多短期異步的小程序。
newFixedThreadPool
固定線程數的池子，每個線程的存活時間是無限的，當池子滿了就不再添加線程；若池中線程均在繁忙狀態，新任務會進入阻塞隊列中（無界的阻塞隊列）。
適用場景：執行長期的任務，性能較好。
newSingleThreadExecutor
只有一個線程的線程池，且線程的存活時間是無限的；當線程繁忙時，對於新任務會進入阻塞隊列中（無界的阻塞隊列）。
適用：一個任務一個任務執行的場景。
NewScheduledThreadPool
創建一個固定大小的線程池，池內的線程存活時間無限，線程池支持定時及週期性的任務執行。如果所有線程均處於繁忙狀態，對於新任務會進入DelayedWorkQueue隊列。
適用場景：週期性執行任務的場景。
線程池任務執行流程：
- 當線程池小於corePoolSize時，新任務將創建一個新的線程，即使此時線程池種存在空閒線程。
- 當線程池達到corePoolSize時，新提交的任務將被放入workQueue中，等待線程池任務調度執行。
- 當workQueue已滿，且maximumPoolSize>corePoolSize時，新任務會創建新線程執行任務。
- 當提交任務數超過maximumPoolSize時，新提交任務由RejectedExecutionHandler處理。
- 當線程池中超過corePoolSize時，空閒時間達到keepAliveTime時，關閉空閒線程。
- 當設置了allowCoreThreadTimeOut(true)時，線程池中corePoolSize線程空閒時間達到keepAliveTime也將關閉。
ThreadPoolExecutor使用場景。以及原理
ThreadPoolExecutor類實現了ExecutorService接口和Executor接口。
ThreadPoolExecutor參數：
參數含義corePoolSize核心線程池大小maximumPoolSize最大線程池大小keepAliveTime線程池中超過corePoolSize數目的空閒線程最大存活時間；可以allowCoreThreadTimeOut(true)使得核心線程有效時間TimeUnitkeepAliveTime時間單位workQueue阻塞任務隊列threadFactory新建線程工廠RejectedExecutionHandler當提交任務數超過maxmumPoolSize+workQueue之和時，任務會交給RejectedExecutionHandler來處理
- 當線程池小於corePoolSize時，新提交任務將創建一個新線程執行任務，即使此時線程池中存在空閒線程。
- 當線程池達到corePoolSize時，新提交任務將被放入workQueue中，等待線程池中任務調度執行
- 當workQueue已滿，且maximumPoolSize>corePoolSize時，新提交任務會創建新線程執行任務
- 當提交任務數超過maximumPoolSize時，新提交任務由RejectedExecutionHandler處理
- 當線程池中超過corePoolSize線程，空閒時間達到keepAliveTime時，關閉空閒線程
- 當設置allowCoreThreadTimeOut(true)時，線程池中corePoolSize線程空閒時間達到keepAliveTime也將關閉
Executor拒絕策略說的是什麼？
線程池中的數量大於corePoolSize，緩衝隊列workQueue滿，並且線程池中的數量等於maximumPoolSize，那麼通過 handler所指定的策略來處理此任務。
ThreadPoolExecutor.AbortPolicy
拋出java.util.concurrent.RejectedExecutionException異常。
ThreadPoolExecutor.CallerRunsPolicy
用於被拒絕任務的處理程序，它直接在 execute 方法的調用線程中運行被拒絕的任務；如果執行程序已關閉，則會丟棄該任務。
ThreadPoolExecutor.DiscardOldestPolicy
丟棄任務隊列中最舊任務。
ThreadPoolExecutor.DiscardPolicy
丟棄當前將要加入隊列的任務。
無界阻塞延遲隊列delayqueue原理是什麼？
DelayQueue是一個支持延時獲取元素的無界阻塞隊列。隊列使用PriorityQueue來實現。隊列中的元素必須實現Delayed接口，在創建元素時可以指定多久才能從隊列中獲取當前元素。只有在延遲期滿時才能從隊列中提取元素。
適用場景
緩存系統的設計：使用DelayQueue保存緩存元素的有效期，使用一個線程循環查詢DelayQueue，一旦能從DelayQueue中獲取元素時，就表示有緩存到期了。
定時任務調度：使用DelayQueue保存當天要執行的任務和執行時間，一旦從DelayQueue中獲取到任務就開始執行，比如Timer就是使用DelayQueue實現的。
實現思路
以支持優先級的PriorityQueue無界隊列作為一個容器，因為元素都必須實現Delayed接口，可以根據元素的過期時間來對元素進行排列，因此，先過期的元素會在隊首，每次從隊列裡取出來都是最先要過期的元素。如果延遲隊列中的消息到了延遲時間則可以從中取出消息否則無法取出消息也就無法消費。
CyclicBarrier和CountDownLatch的區別
CountDownLatchCyclicBarrier減計數方式加計數方式計算為0時釋放所有等待的線程計數達到指定值時釋放所有等待線程計數為0時，無法重置計數達到指定值時，計數置為0重新開始調用countDown()方法計數減一，調用await()方法只進行阻塞，對計數沒任何影響調用await()方法計數加1，若加1後的值不等於構造方法的值，則線程阻塞不可重複利用可重複利用
CyclicBarrier
class Runner implements Runnable {
private CyclicBarrier barrier;
private String name;
public Runner(CyclicBarrier barrier, String name) {
super();
this.barrier = barrier;
this.name = name;
}
@Override
public void run() {
try {
Thread.sleep(1000 * (new Random()).nextInt(8));
System.out.println(name + " 準備OK.");
barrier.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
e.printStackTrace();
}
System.out.println(name + " Go!!");
}
}
public class Race {
public static void main(String[] args) throws IOException, InterruptedException {
CyclicBarrier barrier = new CyclicBarrier(3);
ExecutorService executor = Executors.newFixedThreadPool(3);
executor.submit(new Thread(new Runner(barrier, "zhangsan")));
executor.submit(new Thread(new Runner(barrier, "lisi")));
executor.submit(new Thread(new Runner(barrier, "wangwu")));
executor.shutdown();
}
}
CyclicBarrier就是一個柵欄，等待所有線程到達後再執行相關的操作。barrier 在釋放等待線程後可以重用。
CountDownLatch
public class TestCountDownLatch {
private static final int N = 10;
public static void main(String[] args) throws InterruptedException {
CountDownLatch doneSignal = new CountDownLatch(N);
CountDownLatch startSignal = new CountDownLatch(1);//開始執行信號
for (int i = 1; i <= N; i++) {
new Thread(new Worker(i, doneSignal, startSignal)).start();//線程啟動了
}
System.out.println("begin------------");
startSignal.countDown();//開始執行啦
doneSignal.await();//等待所有的線程執行完畢
System.out.println("Ok");
}
static class Worker implements Runnable {
private final CountDownLatch doneSignal;
private final CountDownLatch startSignal;
private int beginIndex;
Worker(int beginIndex, CountDownLatch doneSignal,
CountDownLatch startSignal) {
this.startSignal = startSignal;
this.beginIndex = beginIndex;
this.doneSignal = doneSignal;
}
public void run() {
try {
startSignal.await(); //等待開始執行信號的發佈
beginIndex = (beginIndex - 1) * 10 + 1;
for (int i = beginIndex; i <= beginIndex + 10; i++) {
System.out.println(i);
}
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
doneSignal.countDown();
}
}
}
}
CountDownLatch 是計數器, 線程完成一個就記一個, 就像 報數一樣, 只不過是遞減的。
而CyclicBarrier更像一個水閘, 線程執行就像水流, 在水閘處都會堵住, 等到水滿(線程到齊)了, 才開始洩流。
閱讀更多 IT曉月寒丶 的文章
