JUC中Future及CompletableFuture的用法與說明
Future 異步回調（不推薦）
new Thread時傳入FutureTask對象（構造時傳入Callable任務對象），調用start啟動
創(chuàng)建線程的方式
1.直接new Thread()對象，重寫run方法，調用start啟動
2.new Thread時傳入Runnable任務對象（重寫run方法），調用start啟動
3.new Thread時傳入FutureTask對象（構造時傳入Callable任務對象），調用start啟動
/**
* 創(chuàng)建線程的三種方式
*/
public class CreateThreadDemo {
public static void main(String[] args) throws Exception {
// 1.重寫Thread的run方法
Thread t1 = new Thread(() -> {
System.out.println("第一種創(chuàng)建線程方式，重寫Thread的run方法");
},"t1");
// 2.重寫Runnable的run方法
RunTask runTask = new RunTask();
Thread t2 = new Thread(runTask, "t2");
// 3.傳入FutureTask對象，重寫Callable的call方法（異步帶返回值）
CallTask callTask = new CallTask();
FutureTask<String> futureTask = new FutureTask<>(callTask);
Thread t3 = new Thread(futureTask, "t3");
t1.start();
t2.start();
t3.start();
System.out.println("異步執(zhí)行結果" + futureTask.get() + " " + System.currentTimeMillis());
}
}
/**
* Runnable 任務
*/
class RunTask implements Runnable {
@Override
public void run() {
System.out.println("runnable running " + System.currentTimeMillis());
}
}
/**
* Callable 任務
*/
class CallTask implements Callable<String> {
@Override
public String call() throws Exception {
System.out.println("callable running " + System.currentTimeMillis());
return "hello";
}
}概述
Fucture接口（FutureTask實現類）定義了操作異步任務執(zhí)行的一些方法，如獲取異步任務的執(zhí)行結果、取消任務的執(zhí)行、判斷任務是否被取消、判斷任務是否完畢等。
Future接口可以為主線程開一個分支任務，專門為主線程處理耗時費力的復雜業(yè)務。
Future提供了一種異步并行計算的功能：如果主線程需要執(zhí)行一個很耗時的計算任務，我們就可以通過future把這個任務放到異步線程中執(zhí)行，主線程繼續(xù)處理其他任務或者先行結束，再通過Future獲取計算結果。
Future接口 --- 源碼
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}FutureTask（Future實現類）
Thread類構造方法：
可以看出Thread構造方法接收的任務對象只有Runnable接口，為什么還能接收FutureTask任務對象呢？
因為FutureTask實現了RunableFuture接口，而RunnableFuture接口繼承了Runnable接口和Future接口，所以FutureTask也是Runnable對象，而FutureTask可以接收Callable任務對象是因為構造方法中提供了接收Callable對象的構造方法。
Future編碼優(yōu)缺點
- 優(yōu)點：future+線程池可以實現多線程任務配合，能顯著提高程序的執(zhí)行效率
- 缺點：get()阻塞、isDone()輪詢導致CPU空轉
- 阻塞情況： 調用FutureTask.get方法時線程會阻塞等待異步結果的獲取
代碼：
FutureTask<String> futureTask = new FutureTask<String>(() -> {
System.out.println("開始執(zhí)行......" + now().getSecond());
// 休眠
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("結束......");
return "FutureTask End";
});
Thread t = new Thread(futureTask);
t.start();
System.out.println("子線程執(zhí)行結果：" + futureTask.get());
System.out.println("--------------------------------");
System.out.println("主線程 -------- 執(zhí)行..... " + now().getSecond());
System.out.println("--------------------------------");運行結果：
可以通過設置：
System.out.println("子線程執(zhí)行結果：" + futureTask.get(2, TimeUnit.SECONDS));
暴力終止程序運行或拋出異常
isDone()輪詢導致CPU空轉：
輪詢的方式會消耗無謂的CPU資源，而且也不見得能及時獲得到計算結果，如果想要異步獲取結果，通常都會以輪詢的方式去獲取結果，盡量不要阻塞。
while (true) {
if (futureTask.isDone()) {
System.out.println("子線程執(zhí)行結果：" + futureTask.get() + now().getSecond());
break;
}else {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("暫停2s");
}
}結論：Future對于結果的獲取不是很友好，只能通過阻塞或者輪詢的方式得到任務的結果。
由此引入CompletableFuture，規(guī)避以上的缺點
需要說明的是對于簡單的業(yè)務使用Future就可以了
CompletableFuture（異步任務編排）
CompletableFuture可以代表一個明確完成的Future，也可能代表一個完成階段(CompletionStage),它支持在計算完成以后觸發(fā)一些函數或執(zhí)行某些動作。
CompletableFuture可以解決的問題？
多個任務前后依賴組合處理：想將多個異步任務的計算結果組合起來，后一個異步任務的計算結果需要前一個異步任務的值；將兩個或多個異步計算合成一個異步計算，這幾個異步計算互相獨立，同時后面這個又依賴前一個計算結果
對計算速度選最快：當Future集合中某個任務最快結束時，返回結果，返回第一名的處理結果。。。。
CompletableFuture對比Future的改進
Future問題：get()阻塞、isDone()輪詢CPU空轉
對于真正的異步處理我們希望是可以通過傳入回調函數，在Future結束時自動調用該回調函數，這樣，我們就不用等待結果。
CompletableFuture提供一種觀察者模式類似的機制，可以讓任務執(zhí)行完成后通知監(jiān)聽的一方。
怎么創(chuàng)建CompletableFuture對象？
創(chuàng)建
一般不建議直接new CompletableFuture
可以采用以下的方式創(chuàng)建CompletableFuture對象：
1.CompletableFuture.runAsync無返回值（指定線程池會用指定的線程池，沒有就會使用默認的線程池）
2. CompletableFuture.supplyAsync有返回值（指定線程池會用指定的線程池，沒有就會使用默認的線程池）
怎么解決Future的阻塞和輪詢問題？
從java8開始引入了CompletableFuture，它是Future的功能增強版，減少阻塞和輪詢，可以傳入回調對象，當異步任務完成或者發(fā)生異常時，自動調用回調對象的回調方法。
默認線程池創(chuàng)建的是守護線程，自定義線程池是用戶線程
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService threadPool = Executors.newFixedThreadPool(5);
// 有返回值
supplyFuture(threadPool);
System.out.println("主線程 running " + now());
threadPool.shutdown();
}
private static void supplyFuture(ExecutorService threadPool) {
CompletableFuture<Integer> supplyFuture = CompletableFuture.supplyAsync(() -> {
System.out.println("開始執(zhí)行...." +
(Thread.currentThread().isDaemon() ? "守護線程" : "用戶線程") +
" " + now());
int random = ThreadLocalRandom.current().nextInt();
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
return random;
}, threadPool).whenComplete((v, e) -> {
if (e == null) {
// 沒有異常
System.out.println("隨機數結果： " + v);
}
}).exceptionally((e) -> {
e.printStackTrace();
System.out.println("異常情況：" + e.getCause());
return null;
});
}運行結果：
已經解決了Future編碼出現的get阻塞問題。
上述代碼塊邏輯解析：
supplyAsync()中是任務（默認線程池創(chuàng)建的守護線程），如果這個任務成功會走到whenComplete代碼塊中，不成功會走到exceptionally代碼塊中
注意：默認線程池創(chuàng)建的是守護線程，自定義線程池是用戶線程，守護線程會隨著用戶線程的結束而結束，所以會導致主線程執(zhí)行完了然后還沒打印出隨機數線程池就關閉了，就是以下輸出的情況
怎么獲得異步結果？
通過get或者join獲得結果（區(qū)別在于join在編譯期間不會作檢查性異常的處理，拋不拋異常都可以）
CompletableFuture的優(yōu)點
- 異步任務結束時，會自動回調某個對象的方法；
- 主線程設置好回調后，不再關心異步任務的執(zhí)行，異步任務之間可以順序執(zhí)行；
- 異步任務出錯時，會自動回調某個對象的方法：
函數式編程
函數式編程：Lambda表達式+Stream流式編程+Chain鏈式調用+Java8函數式編程
如Runnable、Function、Consumer --- BIConsumer、Supplier
常用函數式接口
函數式接口名稱 | 方法名稱 | 參數 | 返回值類型 |
Consumer<T> | accept | T t | void |
Supplier<T> | get | 無 | T |
Function<T, R> | apply | T t | R |
Predicate<T> | test | T t | boolean |
BiConsumer<T, U> | accept | T t, U u | void |
BiFunction<T, U, R> | apply | T t, U u | R |
BiPredicate<T, U> | test | T t, U u | boolean |
UnaryOperator<T> | apply | T t | T |
BinaryOperator<T> | apply | T t1, T t2 | T |
IntConsumer | accept | int value | void |
IntSupplier | getAsInt | 無 | int |
IntFunction<R> | apply | int value | R |
IntPredicate | test | int value | boolean |
鏈式語法
public class ChainDemo {
public static void main(String[] args) {
Student student = new Student();
student.setId(1).setName("karry").setMajor("cs");
}
}
@Data
@AllArgsConstructor
@NoArgsConstructor
@Accessors(chain = true)
class Student{
private Integer id;
private String name;
private String major;
}案例精講-電商網站的比價需求
從“功能”到“性能”
/**
* 電商網站的比價需求
*/
public class CompletableMallDemo {
private static final List<Mall> list = new ArrayList<>();
static {
for (int i = 0; i <= 1000; i ++) {
list.add(new Mall("book" + i));
}
}
/**
* 查詢價格 同步處理
* @param list 平臺列表
* @param bookName 書籍名稱
* @return 查詢所需時間
*/
public static Long getPrice(List<Mall> list, String bookName) {
long start = System.currentTimeMillis();
List<String> prices = list.stream()
.map((mall) -> String.format(bookName + " in %s price is %.2f",
mall.getName(), mall.calcPrice(bookName)))
.collect(Collectors.toList());
long end = System.currentTimeMillis();
System.out.println("用時: " + end);
prices.forEach(System.out::println);
return end - start;
}
/**
* 查詢價格 異步處理
* @param list 平臺
* @param bookName 書籍名稱
* @return 查詢所需
*/
public static Long getPriceByCompletableFuture(List<Mall> list, String bookName) {
long start = System.currentTimeMillis();
ExecutorService threadPool = Executors.newFixedThreadPool(10);
List<String> prices = list.stream()
// 把 mall 映射到 CompletableFuture對象
.map(mall -> CompletableFuture.supplyAsync(() ->
String.format(bookName + " in %s price is %.2f",
mall.getName(), mall.calcPrice(bookName)), threadPool)
).collect(Collectors.toList()).stream()
// CompletableFuture對象 映射到 CompletableFuture.join() [String對象]
.map(CompletableFuture::join).collect(Collectors.toList());
threadPool.shutdown();
long end = System.currentTimeMillis();
System.out.println("用時: " + end);
prices.forEach(System.out::println);
return end - start;
}
public static void main(String[] args) {
System.out.println("差值：" +
(getPrice(list, "mysql") - getPriceByCompletableFuture(list, "mysql")));
}
}
@AllArgsConstructor
@NoArgsConstructor
@Data
class Mall {
/**
* 電商網站名稱
*/
private String name;
/**
* 模擬查詢價格
* @param bookName 書籍名稱
* @return double 價格
*/
public double calcPrice(String bookName) {
return ThreadLocalRandom.current().nextDouble(20, 100) + bookName.charAt(0);
}
}源碼分析
public class CompletableFuture<T> implements Future<T>, CompletionStage<T> {
}CompletableFuture實現了Future接口，還拓展了Future不具備的CompletionStage接口
CompletionStage（完成階段）
CompletionStage代表異步計算過程中的某一個階段，一個階段完成后可能會觸發(fā)另外一個階段。
一個階段的計算執(zhí)行可以時一個Fuction，Consumer或者Runnable。比如：stage.thenApply().thenAccept().thenRun()
CompletableFuture API
不帶Async和帶Async的API區(qū)別：
對于CompletionStage，每個任務階段帶Async的任務可以設定線程池，不設定就會使用默認線程池, 而不帶Async的任務階段會使用CompletableFuture設定的線程池。
thenRun和thenRunAsync的區(qū)別：
thenRun使用的supplyAsync或者runAsync傳入的線程池（不傳入則使用默認線程池---守護線程）thenRunAsync使用的是自己api傳入的線程池，不傳入則使用默認線程池（守護線程）
/**
* thenRunAsync方法
* @param threadPool
*/
public static void testRunAsyncMethod(ExecutorService threadPool) {
CompletableFuture.runAsync(() -> {
System.out.println("step 1 " + Thread.currentThread().getName());
}, threadPool).thenRunAsync(() -> {
System.out.println("step 2 " + Thread.currentThread().getName());
}).thenRunAsync(() -> {
System.out.println("step 3 " + Thread.currentThread().getName());
}, threadPool).thenRunAsync(() -> {
System.out.println("step 4 " + Thread.currentThread().getName());
}).thenRunAsync(() -> {
System.out.println("step 5 " + Thread.currentThread().getName());
}).join();
}
方法 | 觸發(fā)時機 | 輸入參數 | 返回值類型 | 核心作用 |
thenRun | 前序任務正常完成后 | 無（Runnable ） | CompletableFuture<Void> | 執(zhí)行無輸入、無輸出的后續(xù)操作 |
thenAccept | 前序任務正常完成后 | 前序任務的結果（Consumer ） | CompletableFuture<Void> | 消費前序任務的結果，無新輸出 |
thenApply | 前序任務正常完成后 | 前序任務的結果（Function ） | CompletableFuture<U> | 基于前序結果計算新結果，有新輸出 |
handle | 前序任務完成（無論成?。?/p> | 前序結果 + 異常（BiFunction ） | CompletableFuture<U> | 處理前序任務的結果或異常，計算新結果 |
1. 獲得結果和觸發(fā)計算
獲得結果：
- get() : 不見不散
- get(long, TimUnit): 過時不候
- join() : 功能和get類似，但是在編譯期間不拋出受檢查異常
- getNow(valueIfAbsent): 立即獲得當前結果，為空則返回valueIfAbsent的值
public T getNow(T valueIfAbsent) {
Object r;
return ((r = result) == null) ? valueIfAbsent : reportJoin(r);
}觸發(fā)計算：
complete(T value) ：是否打斷get方法立即返回括號值（返回括號值，為true；不返回為false）
public boolean complete(T value) {
boolean triggered = completeValue(value);
postComplete();
return triggered;
}eg:
public static void main(String[] args) throws InterruptedException {
ExecutorService threadPool = Executors.newFixedThreadPool(5);
testCompleteMethodByTrue(threadPool);
testCompleteMethodByFalse(threadPool);
threadPool.shutdown();
}
public static void testCompleteMethodByFalse(ExecutorService threadPool) throws InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "get/join返回的結果";
}, threadPool);
TimeUnit.SECONDS.sleep(4);
boolean flag = future.complete("設定的值");
System.out.println("complete方法返回值為 " + flag + "，4s左右獲得的值為 " + future.join());
}
public static void testCompleteMethodByTrue(ExecutorService threadPool) throws InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "get/join返回的結果";
}, threadPool);
TimeUnit.SECONDS.sleep(2);
boolean flag = future.complete("設定的值");
System.out.println("complete方法返回值為 " + flag + "，2s左右獲得的值為 " + future.join());
}運行結果：
2. 對計算結果進行處理
2.1 thenApply: 計算結果存在依賴關系，串行化
/**
* thenApply
* @param threadPool
*/
public static void testApplyMethod(ExecutorService threadPool) {
StringBuffer str = new StringBuffer();
CompletableFuture<StringBuffer> future = CompletableFuture.supplyAsync(() -> {
return str.append("a");
}, threadPool).thenApply(f -> {
return str.append("b");
}).thenApply(f -> {
return str.append("c");
}).whenComplete((v, e) -> {
if (v != null) {
System.out.println("apply處理后的結果為 " + v);
}
}).exceptionally(e -> {
e.printStackTrace();
return null;
});
}2.2 handle： 計算結果存在依賴關系，串行化（可以帶著）
/**
* handle方法
* @param threadPool
*/
public static void testHandleMethod(ExecutorService threadPool) {
StringBuffer str = new StringBuffer();
CompletableFuture<StringBuffer> future = CompletableFuture.supplyAsync(() -> {
return str.append("a");
}, threadPool).handle((f, e) -> {
if (e == null) {
return f.append("b");
}else {
e.printStackTrace();
return null;
}
}).handle((f, e) -> {
if (e == null) {
return f.append("c");
}else {
e.printStackTrace();
return null;
}
});
System.out.println("handle處理后的結果為 " + future.join());
}
3.對計算結果進行消費
接收任務的處理結果，消費處理，無返回結果
thenAccept:
public CompletableFuture<Void> thenAccept(Consumer<? super T> action) {
return uniAcceptStage(null, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
return uniAcceptStage(asyncPool, action);
}
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action,
Executor executor) {
return uniAcceptStage(screenExecutor(executor), action);
}eg:
/**
* thenAccept方法（消費型）
* @param threadPool
*/
public static void testAcceptMethod(ExecutorService threadPool) {
StringBuffer str = new StringBuffer();
CompletableFuture.supplyAsync(() -> str.append("abc"), threadPool)
.thenAccept(r -> System.out.println("accept直接消費" + r)).join();
}輸出：
thenRun:
任務A執(zhí)行完執(zhí)行B，并且B不需要A的結果
/**
* thenRun方法
* @param threadPool
*/
public static void testRunMethod(ExecutorService threadPool) {
CompletableFuture.runAsync(() -> {
System.out.println("step 1");
}, threadPool).thenRun(() -> {
System.out.println("step 2");
}).thenRun(() -> {
System.out.println("step 3");
});
}4.對計算速度進行選用
applyToEither:
對兩個future對象選用速度較快的那一個結果
eg:
/**
* applyToEither 方法
* @param threadPool
*/
public static void testApplyToEitherMethod(ExecutorService threadPool) {
for (int i = 2; i <= 5; i ++) {
CompletableFuture<String> future = getPlayFuture(i - 1, threadPool).applyToEither(getPlayFuture(i, threadPool), f -> f + " is winner");
System.out.println(future.join());
}
}
private static CompletableFuture<String> getPlayFuture(int num, ExecutorService threadPool) {
return CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.MILLISECONDS.sleep(num * 100);
} catch (InterruptedException e) {
e.printStackTrace();
}
return "play" + num;
});
}
5.對計算結果進行合并
thenCombine:
兩個CompletionStage任務都完成后，最終能把兩個任務的結果一起提交給thenCombine來處理；先完成的先等著，等待其它分支任務。
eg：
/**
* thenCombine 方法
* @param threadPool
*/
public static void testCombineMethod(ExecutorService threadPool) {
CompletableFuture<Integer> future1 = getBranchFuture(1, threadPool);
CompletableFuture<Integer> future2 = getBranchFuture(2, threadPool);
CompletableFuture<Integer> combineFuture = future2.thenCombine(future1, Integer::sum);
System.out.println("計算結果為 " + combineFuture.join());
}
private static CompletableFuture<Integer> getBranchFuture(int num, ExecutorService threadPool) {
return CompletableFuture.supplyAsync(() -> {
try {
TimeUnit.MILLISECONDS.sleep(num * 100);
} catch (InterruptedException e) {
e.printStackTrace();
}
return num * 10;
});
}輸出：計算結果為 30
總結
以上為個人經驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
Java中BigDecimal與0比較的一個坑實戰(zhàn)記錄
BigDecimal屬于大數據,精度極高,不屬于基本數據類型,屬于java對象,下面這篇文章主要給大家介紹了關于Java中BigDecimal與0比較的一個坑的相關資料,需要的朋友可以參考下2022-12-12
這篇文章主要為大家詳細介紹了基于java語言實現快遞系統(tǒng)，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-03-03
這篇文章主要為大家介紹了Springboot核心知識點中的數據訪問配置，文中的示例代碼講解詳細，對我們了解SpringBoot有一定幫助，快跟隨小編一起學習一下吧2021-12-12
SpringBoot項目多模塊項目中父類與子類pom.xml的關聯(lián)問題小結
這篇文章主要介紹了SpringBoot項目多模塊項目中父類與子類pom.xml的關聯(lián)問題,本文通過實例代碼給大家介紹的非常詳細,對大家的學習或工作具有一定的參考借鑒價值,需要的朋友參考下吧2025-04-04
String類下compareTo()與compare()方法比較
這篇文章主要介紹了String類下compareTo()與compare()方法比較的相關資料,需要的朋友可以參考下2017-05-05
排序算法看似簡單，其實不同的算法中蘊涵著經典的算法策略。通過熟練掌握排序算法，就可以掌握基本的算法設計思想，本文主要介紹了Java中的排序算法，需要的朋友歡迎閱讀2023-04-04
這篇文章主要介紹了jackson 實現將實體轉json json字符串轉實體，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-10-10
Mybatis-Plus(簡稱MP)是一個Mybatis的增強工具,在Mybatis的基礎上只做增強不做改變,為簡化開發(fā)、提高效率而生,下面這篇文章主要給大家介紹了關于MybatisPlus搭建項目環(huán)境及分頁插件的相關資料,需要的朋友可以參考下2022-11-11
這篇文章主要介紹了springBoot2.X配置全局捕獲異常的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-07-07
這篇文章主要介紹了MyBatis 接收數據庫中沒有的字段的解決方案，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-03-03