1??J.U.C-FutureTask
1 概述
FutureTask這個組件是JUC里邊的但是它不是AQS的子類;創(chuàng)建線程常用的方式有兩種一種是繼承Thread,一種是實現(xiàn)Runnable接口,這兩種方式有一個共同的缺陷就是執(zhí)行完任務(wù)以后無法獲取執(zhí)行結(jié)果,從JDK1.5開始就提供了Future和Callable通過他們可以在任務(wù)執(zhí)行完畢以后獲取結(jié)果;
2 Callable Future FutureTask的使用方法
① Callable與Runnable接口的對比
Runnable的代碼非常簡單,它是一個接口而且只有一個方法run();創(chuàng)建一個類實現(xiàn)它并在里邊寫一些操作,然后使用線程去執(zhí)行該Runnable實現(xiàn)類即可以實現(xiàn)多線程;
②Callable的代碼也非常簡單,不同的是它是一個泛型的接口它里邊有一個call()方法,call()方法的返回值類型就是我們創(chuàng)建Callable時傳進(jìn)去的V類型,Callable與Runnable的功能大致相似;相比較而言Callable比Runnable稍微強(qiáng)大一些,因為Callable對線程執(zhí)行后可以有返回值并且可以拋出異常;
③Future也是一個接口,Future可以取消任務(wù)查詢?nèi)蝿?wù)是否被取消以及獲取結(jié)果等等操作;通常線程都是異步計算模型的,通常我們不能直接從別的線程中得到方法的返回值這個時候我們就可以使用Future來完成;Future可以監(jiān)視目標(biāo)線程調(diào)用call()的情況,當(dāng)我們調(diào)用Future的get()方法的時候就可以獲取到結(jié)果,這個時候通常線程可能不會直接完成,當(dāng)前線程就會開始阻塞,直到call()方法結(jié)束返回結(jié)果線程才會繼續(xù)執(zhí)行,Future可以得到別的線程任務(wù)的方法的返回值.
④ FutureTask它的父類是RunnableFuture,而RunnableFuture繼承了Runnable與Future這兩個接口,由此我們可以直到FutureTask最終也是執(zhí)行
Callable類型的任務(wù),如果構(gòu)造函數(shù)參數(shù)是Runnable的話,它會轉(zhuǎn)換成Callable類型,所以FutureTask即可以作為Runnable被線程執(zhí)行也可以作為Future拿到線程的執(zhí)行結(jié)果;
3 代碼演示
// Future演示
@Slf4j
public class FutureExample {
static class MyCallable implements Callable<String> {
@Override
public String call() throws Exception {
log.info("do something in callable");
Thread.sleep(5000);
return "Done";
}
}
public static void main(String[] args) throws Exception {
ExecutorService executorService = Executors.newCachedThreadPool();
Future<String> future = executorService.submit(new MyCallable());
log.info("do something in main");
Thread.sleep(1000);
String result = future.get();
log.info("result：{}", result);
}
}
// FutureTask代碼演示
@Slf4j
public class FutureTaskExample {
public static void main(String[] args) throws Exception {
FutureTask<String> futureTask = new FutureTask<String>(new Callable<String>() {
@Override
public String call() throws Exception {
log.info("do something in callable");
Thread.sleep(5000);
return "Done";
}
});
new Thread(futureTask).start();
log.info("do something in main");
Thread.sleep(1000);
String result = futureTask.get();
log.info("result：{}", result);
}
}
2??J.U.C-ForkJoin
1 概述
ForkJoin框架是JDK7提供的一個用于并行執(zhí)行任務(wù)的框架,他將一個大任務(wù)分割成若干個小任務(wù),最終匯總每個小任務(wù)的結(jié)果后得到大任務(wù)的結(jié)果;從字面上看Fork就是切割任務(wù),Join就是合并結(jié)果并得到最終的結(jié)果;它主要采取的是工作竊取算法,工作竊取算法主要是指某個線程從其他隊列里竊取任務(wù)來執(zhí)行;
2 工作竊取流程圖
3 為什么要使用工作竊取算法?
原理 :
假如我們需要做一個比較大的任務(wù),我們可以把這個任務(wù)分割成互不依賴的子任務(wù),為了減少線程間的競爭,于是把這些子任務(wù)分別放到不同的隊列里,為每一個隊列創(chuàng)建一個單獨的線程來執(zhí)行隊列里的任務(wù),線程和隊列一一對應(yīng),比如A線程負(fù)責(zé)處理A線程里的任務(wù),但是有一些線程會先于其他線程完成任務(wù),這個時候其他線程對應(yīng)的隊列里還有任務(wù)等待處理,此時就需要使用工作竊取來讓空閑的線程獲取別的線程的任務(wù);但是這個時候多個線程會訪問同一個隊列,為了減少竊取任務(wù)線程與被竊取任務(wù)線程之間競爭通常我們會使用雙端隊列,被竊取任務(wù)的線程永遠(yuǎn)從雙端隊列的頭部拿任務(wù)進(jìn)行處理,而竊取任務(wù)的線程從雙端隊列的底部獲取任務(wù)執(zhí)行;
優(yōu)點 :
這個工作竊取算法的優(yōu)點就是充分利用線程進(jìn)行并行計算并減少了線程間的競爭;
缺點 :
它的缺點是在個別情況下還是會存在競爭(比如雙端隊列里只有一個任務(wù)時),同時這樣也消耗了更多的系統(tǒng)資源(創(chuàng)建了多個線程和多個雙端隊列)
工作竊取算法在ForkJoin中的應(yīng)用 :
對于ForkJoin框架而言,當(dāng)一個任務(wù)正在等待它使用Join操作創(chuàng)建的子任務(wù)結(jié)束時,執(zhí)行這個任務(wù)的工作線程查找其他未被執(zhí)行的任務(wù)并開始執(zhí)行,通過這種方式線程充分利用他們的運(yùn)行時間來提高應(yīng)用系統(tǒng)的性能,
ForkJoin的局限性 :
由于ForkJoin框架使用了工作竊取算法所以執(zhí)行的任務(wù)有一些局限性;
① 首先任務(wù)只能使用Fork和Join操作來實現(xiàn)同步機(jī)制,如果使用了其他同步機(jī)制,他們在工作時工作線程就不能執(zhí)行其他任務(wù)了;比如在ForkJoin框架中使任務(wù)進(jìn)入休眠狀態(tài),那么在休眠期間正在執(zhí)行的工作線程就不會執(zhí)行其他任務(wù)了;
② 我們拆分的任務(wù)不應(yīng)該去執(zhí)行IO操作(如讀或者寫數(shù)據(jù)文件);
③ 任務(wù)不能拋出檢查異常,必須通過必要的代碼來處理他們;
ForkJoin框架的核心 :
ForkJoinPool : 負(fù)責(zé)做實現(xiàn)(包括實現(xiàn)工作竊取算法),它管理工作線程和提供關(guān)于任務(wù)的狀態(tài)以及執(zhí)行信息;
ForkJoinTask : 則主要提供在任務(wù)中執(zhí)行Fork和Join操作的機(jī)制;
4 ForkJoin代碼實例
// ForkJoinTask代碼演示
@Slf4j
public class ForkJoinTaskExample extends RecursiveTask<Integer> {
public static final int threshold = 2;
private int start;
private int end;
public ForkJoinTaskExample(int start, int end) {
this.start = start;
this.end = end;
}
@Override
protected Integer compute() {
int sum = 0;
//如果任務(wù)足夠小就計算任務(wù)
boolean canCompute = (end - start) <= threshold;
if (canCompute) {
for (int i = start; i <= end; i++) {
sum += i;
}
} else {
// 如果任務(wù)大于閾值，就分裂成兩個子任務(wù)計算
int middle = (start + end) / 2;
ForkJoinTaskExample leftTask = new ForkJoinTaskExample(start, middle);
ForkJoinTaskExample rightTask = new ForkJoinTaskExample(middle + 1, end);
// 執(zhí)行子任務(wù)
leftTask.fork();
rightTask.fork();
// 等待任務(wù)執(zhí)行結(jié)束合并其結(jié)果
int leftResult = leftTask.join();
int rightResult = rightTask.join();
// 合并子任務(wù)
sum = leftResult + rightResult;
}
return sum;
}
public static void main(String[] args) {
ForkJoinPool forkjoinPool = new ForkJoinPool();
//生成一個計算任務(wù)，計算1+2+3+4
ForkJoinTaskExample task = new ForkJoinTaskExample(1, 100);
//執(zhí)行一個任務(wù)
Future<Integer> result = forkjoinPool.submit(task);
try {
log.info("result:{}", result.get());
} catch (Exception e) {
log.error("exception", e);
}
}
}
3??J.U.C-BlockingQueue
1 概述
BlockingQueue意思是阻塞隊列,從阻塞這個詞我們就可以看出來在某些情況下對阻塞隊列的訪問可能會造成阻塞;
2 被阻塞的情況
① 當(dāng)隊列已經(jīng)滿了的情況下的入隊列操作(除非有另一個線程進(jìn)行了出隊列操作);
② 當(dāng)隊列已經(jīng)空了的情況下的出隊列操作(除非有另一個線程進(jìn)行了如隊列操作);
3 適用場景
通過上邊的介紹以及阻塞隊列的特性我們可以得出阻塞隊列是線程安全的,它主要使用在生產(chǎn)者消費(fèi)者的場景;
4 阻塞隊列的常用方法
它提供了4套方法,大家可以根據(jù)自己的實際場景來選擇;
5 阻塞隊列的實現(xiàn)類
① ArrayBlockingQueue : ArrayBlockingQueue是一個有界的阻塞隊列,它的內(nèi)部實現(xiàn)是一個數(shù)組,它的容量是有限的,我們必須在初始化的時候指定容量的大小且指定的大小值在指定以后就不能在更改,它是以先進(jìn)先出的方式存儲數(shù)據(jù)的,最新插入的對象是尾部最新移除的對象是頭部;
② DelayQueue : DelayQueue阻塞的是內(nèi)部元素,DelayQueue中的元素必須實現(xiàn)Delayed接口,Delayed接口繼承了Comparable接口(因為DelayQueue中的元素需要進(jìn)行排序),一般情況下都是按照元素的過期時間優(yōu)先級進(jìn)行排序,DelayQueue的應(yīng)用場景比如定義關(guān)閉連接,緩存對象,超時處理等多種場景,它的內(nèi)部實現(xiàn)用的是lock(鎖)與PriorityQueue(排序)
③ LinkedBlockingQueue : 它的大小是可選的,如果初始化時指定了大小那么它就是有邊界的如果不指定就是無邊界的(其實使用的默認(rèn)的最大的整型值),它內(nèi)部實現(xiàn)是一個鏈表,處理底層的結(jié)構(gòu)不一樣其他的都與ArrayBlockingQueue一樣;它也是以先進(jìn)先出的方式存儲數(shù)據(jù),最新插入的對象在尾部最新移除的對象在頭部,
④ PriorityBlockingQueue : 它是一個帶有優(yōu)先級的阻塞隊列且它是一個沒有邊界的隊列但是它是有排序規(guī)則的,PriorityBlockingQueue是允許插入null的,在使用的PriorityBlockingQueue的時候需要注意,所以插入的對象必須實現(xiàn)Compilable接口,隊列優(yōu)先級的排序規(guī)則就是按照我們對這個接口的實現(xiàn)來定義的,我們可以從PriorityBlockingQueue獲取一個迭代器,但是這個迭代器并不保證按照我們的迭代器進(jìn)行迭代;
⑤ SynchronousQueue : 這個隊列內(nèi)部僅允許容納一個元素,當(dāng)一個線程插入一個元素后就會被阻塞,除非這個元素被另一個線程消費(fèi),因此我們又稱之為同步隊列,它是一個無界非緩存的隊列準(zhǔn)確的說它不存儲元素,放入的元素之后等待取走以后才能繼續(xù)放入;
⑥ BlockingQueue總結(jié)
它不僅實現(xiàn)了一個完整隊列所具有的基本功能同時在多線程環(huán)境下還自動了管理多線程間的自動等待 喚醒功能,從而使開發(fā)人員可以忽略這些細(xì)節(jié)關(guān)注更高級的功能;
