建站
咨詢
售后
建站咨詢
新聞中心
本文轉(zhuǎn)載自微信公眾號(hào)「月伴飛魚」，作者日常加油站。轉(zhuǎn)載本文請(qǐng)聯(lián)系月伴飛魚公眾號(hào)。
網(wǎng)站設(shè)計(jì)制作過程拒絕使用模板建站；使用PHP+MYSQL原生開發(fā)可交付網(wǎng)站源代碼;符合網(wǎng)站優(yōu)化排名的后臺(tái)管理系統(tǒng)；成都網(wǎng)站制作、做網(wǎng)站、外貿(mào)營銷網(wǎng)站建設(shè)收費(fèi)合理；免費(fèi)進(jìn)行網(wǎng)站備案等企業(yè)網(wǎng)站建設(shè)一條龍服務(wù).我們是一家持續(xù)穩(wěn)定運(yùn)營了十年的成都創(chuàng)新互聯(lián)公司網(wǎng)站建設(shè)公司。
前言
Java 5在concurrency包中引入了java.util.concurrent.Callable 接口，它和Runnable接口很相似，但它可以返回一個(gè)對(duì)象或者拋出一個(gè)異常。
Callable接口使用泛型去定義它的返回類型。Executors類提供了一些有用的方法在線程池中執(zhí)行Callable內(nèi)的任務(wù)。由于Callable任務(wù)是并行的，我們必須等待它返回的結(jié)果。而線程是屬于異步計(jì)算模型，所以不可能直接從別的線程中得到函數(shù)返回值。
java.util.concurrent.Future對(duì)象為我們解決了這個(gè)問題。在線程池提交Callable任務(wù)后返回了一個(gè)Future對(duì)象，使用它可以知道Callable任務(wù)的狀態(tài)和得到Callable返回的執(zhí)行結(jié)果。Future提供了get()方法讓我們可以等待Callable結(jié)束并獲取它的執(zhí)行結(jié)果。
Future的作用
當(dāng)做一定運(yùn)算的時(shí)候，運(yùn)算過程可能比較耗時(shí)，有時(shí)會(huì)去查數(shù)據(jù)庫，或是繁重的計(jì)算，比如壓縮、加密等，在這種情況下，如果我們一直在原地等待方法返回，顯然是不明智的，整體程序的運(yùn)行效率會(huì)大大降低。
我們可以把運(yùn)算的過程放到子線程去執(zhí)行，再通過 Future 去控制子線程執(zhí)行的計(jì)算過程，最后獲取到計(jì)算結(jié)果。
這樣一來就可以把整個(gè)程序的運(yùn)行效率提高，是一種異步的思想。
同時(shí)在JDK 1.8的doc中，對(duì)Future的描述如下：
A Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the result of the computation.
大概意思就是Future是一個(gè)用于異步計(jì)算的接口。
舉個(gè)例子：
比如去吃早點(diǎn)時(shí)，點(diǎn)了包子和涼菜，包子需要等3分鐘，涼菜只需1分鐘，如果是串行的一個(gè)執(zhí)行，在吃上早點(diǎn)的時(shí)候需要等待4分鐘，但是如果你在準(zhǔn)備包子的時(shí)候，可以同時(shí)準(zhǔn)備涼菜，這樣只需要等待3分鐘。
Future就是后面這種執(zhí)行模式。
創(chuàng)建Future
線程池
- class Task implements Callable
{ - public String call() throws Exception {
- return longTimeCalculation();
- }
- }
- ExecutorService executor = Executors.newFixedThreadPool(4);
- // 定義任務(wù):
- Callable
task = new Task(); - // 提交任務(wù)并獲得Future:
- Future
future = executor.submit(task); - // 從Future獲取異步執(zhí)行返回的結(jié)果:
- String result = future.get(); // 可能阻塞
當(dāng)我們提交一個(gè)Callable任務(wù)后，我們會(huì)同時(shí)獲得一個(gè)Future對(duì)象，然后，我們在主線程某個(gè)時(shí)刻調(diào)用Future對(duì)象的get()方法，就可以獲得異步執(zhí)行的結(jié)果。
在調(diào)用get()時(shí)，如果異步任務(wù)已經(jīng)完成，我們就直接獲得結(jié)果。如果異步任務(wù)還沒有完成，那么get()會(huì)阻塞，直到任務(wù)完成后才返回結(jié)果
FutureTask
除了用線程池的 submit 方法會(huì)返回一個(gè) future 對(duì)象之外，同樣還可以用 FutureTask 來獲取 Future 類和任務(wù)的結(jié)果。
我們來看一下 FutureTask 的代碼實(shí)現(xiàn)：
- public class FutureTask
implements RunnableFuture { - ...
- }
可以看到，它實(shí)現(xiàn)了一個(gè)接口，這個(gè)接口叫作 RunnableFuture。
我們再來看一下 RunnableFuture 接口的代碼實(shí)現(xiàn)：
- public interface RunnableFuture
extends Runnable, Future { - void run();
- }
既然 RunnableFuture 繼承了 Runnable 接口和 Future 接口，而 FutureTask 又實(shí)現(xiàn)了 RunnableFuture 接口，所以 FutureTask 既可以作為 Runnable 被線程執(zhí)行，又可以作為 Future 得到 Callable 的返回值。
典型用法是，把 Callable 實(shí)例當(dāng)作 FutureTask 構(gòu)造函數(shù)的參數(shù)，生成 FutureTask 的對(duì)象，然后把這個(gè)對(duì)象當(dāng)作一個(gè) Runnable 對(duì)象，放到線程池中或另起線程去執(zhí)行，最后還可以通過 FutureTask 獲取任務(wù)執(zhí)行的結(jié)果。
下面我們就用代碼來演示一下：
- public class FutureTaskDemo {
- public static void main(String[] args) {
- Task task = new Task();
- FutureTask
integerFutureTask = new FutureTask<>(task); - new Thread(integerFutureTask).start();
- try {
- System.out.println("task運(yùn)行結(jié)果："+integerFutureTask.get());
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException e) {
- e.printStackTrace();
- }
- }
- }
- class Task implements Callable
{ - @Override
- public Integer call() throws Exception {
- System.out.println("子線程正在計(jì)算");
- int sum = 0;
- for (int i = 0; i < 100; i++) {
- sum += i;
- }
- return sum;
- }
- }
在這段代碼中可以看出，首先創(chuàng)建了一個(gè)實(shí)現(xiàn)了 Callable 接口的 Task，然后把這個(gè) Task 實(shí)例傳入到 FutureTask 的構(gòu)造函數(shù)中去，創(chuàng)建了一個(gè) FutureTask 實(shí)例，并且把這個(gè)實(shí)例當(dāng)作一個(gè) Runnable 放到 new Thread() 中去執(zhí)行，最后再用 FutureTask 的 get 得到結(jié)果，并打印出來。
Future常用方法
| 方法名 | 返回值 | 入?yún)?/th> | 備注 | 總結(jié) |
|---|---|---|---|---|
| cancel | boolean | （boolean mayInterruptIfRunning） | 用來取消任務(wù)，如果取消任務(wù)成功則返回true，如果取消任務(wù)失敗則返回false。 | 也就是說Future提供了三種功能：判斷任務(wù)是否完成，能夠中斷任務(wù)，能夠獲取任務(wù)執(zhí)行結(jié)果 |
| isCancelled | boolean | 無 | 方法表示任務(wù)是否被取消成功，如果在任務(wù)正常完成前被取消成功，則返回 true。 | |
| isDone | boolean | 無 | 方法表示任務(wù)是否已經(jīng)完成，若任務(wù)完成，則返回true； | |
| get | V | 無 | 方法用來獲取執(zhí)行結(jié)果，這個(gè)方法會(huì)產(chǎn)生阻塞，會(huì)一直等到任務(wù)執(zhí)行完畢才返回 | |
| get | V | （long timeout, TimeUnit unit） | 用來獲取執(zhí)行結(jié)果，如果在指定時(shí)間內(nèi)，還沒獲取到結(jié)果，就直接返回null |
get()方法
get方法最主要的作用就是獲取任務(wù)執(zhí)行的結(jié)果
我們來看一個(gè)代碼示例：
- public class FutureTest {
- public static void main(String[] args) {
- ExecutorService service = Executors.newFixedThreadPool(10);
- Future
future = service.submit(new CallableTask()); - try {
- System.out.println(future.get());
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException e) {
- e.printStackTrace();
- }
- service.shutdown();
- }
- static class CallableTask implements Callable
{ - @Override
- public Integer call() throws Exception {
- Thread.sleep(3000);
- return new Random().nextInt();
- }
- }
- }
在這段代碼中，main 方法新建了一個(gè) 10 個(gè)線程的線程池，并且用 submit 方法把一個(gè)任務(wù)提交進(jìn)去。
這個(gè)任務(wù)它所做的內(nèi)容就是先休眠三秒鐘，然后返回一個(gè)隨機(jī)數(shù)。
接下來我們就直接把future.get結(jié)果打印出來，其結(jié)果是正常打印出一個(gè)隨機(jī)數(shù)，比如 9527 等。
isDone()方法
該方法是用來判斷當(dāng)前這個(gè)任務(wù)是否執(zhí)行完畢了。
需要注意的是，這個(gè)方法如果返回 true 則代表執(zhí)行完成了;如果返回 false 則代表還沒完成。
但這里如果返回 true，并不代表這個(gè)任務(wù)是成功執(zhí)行的，比如說任務(wù)執(zhí)行到一半拋出了異常。那么在這種情況下，對(duì)于這個(gè) isDone 方法而言，它其實(shí)也是會(huì)返回 true 的，因?yàn)閷?duì)它來說，雖然有異常發(fā)生了，但是這個(gè)任務(wù)在未來也不會(huì)再被執(zhí)行，它確實(shí)已經(jīng)執(zhí)行完畢了。
所以 isDone 方法在返回 true 的時(shí)候，不代表這個(gè)任務(wù)是成功執(zhí)行的，只代表它執(zhí)行完畢了。
我們用一個(gè)代碼示例來看一看，代碼如下所示：
- public class GetException {
- public static void main(String[] args) {
- ExecutorService service = Executors.newFixedThreadPool(20);
- Future
future = service.submit(new CallableTask()); - try {
- for (int i = 0; i < 5; i++) {
- System.out.println(i);
- Thread.sleep(500);
- }
- System.out.println(future.isDone());
- future.get();
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException e) {
- e.printStackTrace();
- }
- }
- static class CallableTask implements Callable
{ - @Override
- public Integer call() throws Exception {
- throw new IllegalArgumentException("Callable拋出異常");
- }
- }
- }
在這段代碼中，可以看到有一個(gè)線程池，并且往線程池中去提交任務(wù)，這個(gè)任務(wù)會(huì)直接拋出一個(gè)異常。
那么接下來我們就用一個(gè) for 循環(huán)去休眠，同時(shí)讓它慢慢打印出 0 ~ 4 這 5 個(gè)數(shù)字，這樣做的目的是起到了一定的延遲作用。
在這個(gè)執(zhí)行完畢之后，再去調(diào)用 isDone() 方法，并且把這個(gè)結(jié)果打印出來，然后再去調(diào)用 future.get()
cancel方法
如果不想執(zhí)行某個(gè)任務(wù)了，則可以使用 cancel 方法，會(huì)有以下三種情況：
第一種情況最簡單，那就是當(dāng)任務(wù)還沒有開始執(zhí)行時(shí)，一旦調(diào)用 cancel，這個(gè)任務(wù)就會(huì)被正常取消，未來也不會(huì)被執(zhí)行，那么 cancel 方法返回 true。
第二種情況也比較簡單。如果任務(wù)已經(jīng)完成，或者之前已經(jīng)被取消過了，那么執(zhí)行 cancel 方法則代表取消失敗，返回 false。因?yàn)槿蝿?wù)無論是已完成還是已經(jīng)被取消過了，都不能再被取消了。
第三種情況就是這個(gè)任務(wù)正在執(zhí)行，這個(gè)時(shí)候會(huì)根據(jù)我們傳入的參數(shù)mayInterruptIfRunning做判斷，如果傳入的參數(shù)是 true，執(zhí)行任務(wù)的線程就會(huì)收到一個(gè)中斷的信號(hào)，正在執(zhí)行的任務(wù)可能會(huì)有一些處理中斷的邏輯，進(jìn)而停止，如果傳入的是 false 則就代表不中斷正在運(yùn)行的任務(wù)
isCancelled()方法
判斷是否被取消，它和 cancel 方法配合使用，比較簡單。
應(yīng)用場景
目前對(duì)于Future方式，我們經(jīng)常使用的有這么幾類：
Guava
ListenableFutrue，通過增加監(jiān)聽器的方式，計(jì)算完成時(shí)立即得到結(jié)果，而無需一直循環(huán)查詢
CompletableFuture
Java8的CompletableFuture，使用thenApply，thenApplyAsync可以達(dá)到和Guava類似的鏈?zhǔn)秸{(diào)用效果。
不同的是，對(duì)于Java8，如果thenApplyAsync不傳入線程池，則會(huì)使用ForkJoinPools線程池來執(zhí)行對(duì)應(yīng)的方法，如此可以避免對(duì)其他線程產(chǎn)生影響。
Netty
Netty解決的問題：
- 原生Future的isDone()方法判斷一個(gè)異步操作是否完成，但是定義比較模糊：正常終止、拋出異常、用戶取消都會(huì)使isDone方法返回true。
- 對(duì)于一個(gè)異步操作，我們有些時(shí)候更關(guān)注的是這個(gè)異步操作觸發(fā)或者結(jié)束后能否再執(zhí)行一系列的動(dòng)作。
與JDK相比，增加了完成狀態(tài)的細(xì)分，增加了監(jiān)聽者，異步線程結(jié)束之后能夠觸發(fā)一系列的動(dòng)作。
注意事項(xiàng)
添加超時(shí)機(jī)制
假設(shè)一共有四個(gè)任務(wù)需要執(zhí)行，我們都把它放到線程池中，然后它獲取的時(shí)候是按照從 1 到 4 的順序，也就是執(zhí)行 get() 方法來獲取的
代碼如下所示：
- public class FutureDemo {
- public static void main(String[] args) {
- //創(chuàng)建線程池
- ExecutorService service = Executors.newFixedThreadPool(10);
- //提交任務(wù)，并用 Future 接收返回結(jié)果
- ArrayList
allFutures = new ArrayList<>(); - for (int i = 0; i < 4; i++) {
- Future
future; - if (i == 0 || i == 1) {
- future = service.submit(new SlowTask());
- } else {
- future = service.submit(new FastTask());
- }
- allFutures.add(future);
- }
- for (int i = 0; i < 4; i++) {
- Future
future = allFutures.get(i); - try {
- String result = future.get();
- System.out.println(result);
- } catch (InterruptedException e) {
- e.printStackTrace();
- } catch (ExecutionException e) {
- e.printStackTrace();
- }
- }
- service.shutdown();
- }
- static class SlowTask implements Callable
{ - @Override
- public String call() throws Exception {
- Thread.sleep(5000);
- return "速度慢的任務(wù)";
- }
- }
- static class FastTask implements Callable
{ - @Override
- public String call() throws Exception {
- return "速度快的任務(wù)";
- }
- }
- }
可以看出，在代碼中我們新建了線程池，并且用一個(gè) list 來保存 4 個(gè) Future。
其中，前兩個(gè) Future 所對(duì)應(yīng)的任務(wù)是慢任務(wù)，也就是代碼下方的 SlowTask，而后兩個(gè) Future 對(duì)應(yīng)的任務(wù)是快任務(wù)。
慢任務(wù)在執(zhí)行的時(shí)候需要 5 秒鐘的時(shí)間才能執(zhí)行完畢，而快任務(wù)很快就可以執(zhí)行完畢，幾乎不花費(fèi)時(shí)間。
在提交完這 4 個(gè)任務(wù)之后，我們用 for 循環(huán)對(duì)它們依次執(zhí)行 get 方法，來獲取它們的執(zhí)行結(jié)果，然后再把這個(gè)結(jié)果打印出來。
實(shí)際上在執(zhí)行的時(shí)候會(huì)先等待 5 秒，然后再很快打印出這 4 行語句。
所以問題是：
第三個(gè)的任務(wù)量是比較小的，它可以很快返回結(jié)果，緊接著第四個(gè)任務(wù)也會(huì)返回結(jié)果。
但是由于前兩個(gè)任務(wù)速度很慢，所以我們在利用 get 方法執(zhí)行時(shí)，會(huì)卡在第一個(gè)任務(wù)上。也就是說，雖然此時(shí)第三個(gè)和第四個(gè)任務(wù)很早就得到結(jié)果了，但我們在此時(shí)使用這種 for 循環(huán)的方式去獲取結(jié)果，依然無法及時(shí)獲取到第三個(gè)和第四個(gè)任務(wù)的結(jié)果。直到 5 秒后，第一個(gè)任務(wù)出結(jié)果了，我們才能獲取到，緊接著也可以獲取到第二個(gè)任務(wù)的結(jié)果，然后才輪到第三、第四個(gè)任務(wù)。
假設(shè)由于網(wǎng)絡(luò)原因，第一個(gè)任務(wù)可能長達(dá) 1 分鐘都沒辦法返回結(jié)果，那么這個(gè)時(shí)候，我們的主線程會(huì)一直卡著，影響了程序的運(yùn)行效率。
此時(shí)我們就可以用 Future 的帶超時(shí)參數(shù)的get(long timeout, TimeUnit unit)方法來解決這個(gè)問題。
這個(gè)方法的作用是，如果在限定的時(shí)間內(nèi)沒能返回結(jié)果的話，那么便會(huì)拋出一個(gè) TimeoutException 異常，隨后就可以把這個(gè)異常捕獲住，或者是再往上拋出去，這樣就不會(huì)一直卡著了。
源碼分析
超時(shí)實(shí)現(xiàn)原理
具體實(shí)現(xiàn)類：FutureTask
get()方法可以分為兩步：
- 判斷當(dāng)前任務(wù)的執(zhí)行狀態(tài)，如果不是COMPLETING，就調(diào)用awaitDone()方法開始進(jìn)行死循環(huán)輪旋，如果任務(wù)還沒有執(zhí)行完成會(huì)使用nanos = deadline - System.nanoTime()檢查是否超時(shí)，如果方法已經(jīng)超時(shí)，則會(huì)返回，在返回后如果任務(wù)的狀態(tài)仍然<=COMPLETING，就會(huì)拋出TimeoutException()。
- 如果調(diào)用時(shí)任務(wù)沒有執(zhí)行完成，會(huì)調(diào)用parkNanos()，調(diào)用線程會(huì)阻塞在這里。
接下來分兩種情況：
- 在阻塞時(shí)間完以后任務(wù)的執(zhí)行狀態(tài)仍然沒有改變?yōu)橥瓿桑M(jìn)入下一次循環(huán)，直接返回。
- 如果在輪詢中狀態(tài)已經(jīng)改變，任務(wù)完成，則會(huì)中斷死循環(huán)，返回任務(wù)執(zhí)行的返回值。
本文標(biāo)題：關(guān)于并發(fā)編程，你必須要知道的Future機(jī)制！
鏈接分享：http://m.fisionsoft.com.cn/article/dhjcejg.html
