解讀CompletableFuture的底層原理
引言
在現代 Java 編程中，異步編程變得越來越重要。為了實現高效和非阻塞的代碼，Java 8 引入了 CompletableFuture，一個用于構建異步應用程序的強大工具。
本文將詳細探討 CompletableFuture 的底層原理，展示其工作機制，并通過代碼示例說明如何在實際應用中使用它。
異步編程的背景
異步編程是指在程序運行過程中，不等待某個操作完成，而是繼續(xù)執(zhí)行其他操作，待異步操作完成后再處理其結果。這樣可以提高程序的效率，特別是在 I/O 操作和網絡請求等耗時操作中。
在 Java 8 之前，實現異步編程主要依賴于 Future 接口。然而，Future 存在一些局限性，例如無法手動完成、不能鏈式調用等。為了解決這些問題，Java 8 引入了 CompletableFuture。
什么是 CompletableFuture
CompletableFuture 是 Java 8 中新增的類，實現了 Future 和 CompletionStage 接口，提供了強大的異步編程能力。
CompletableFuture 允許以非阻塞的方式執(zhí)行任務，并且可以通過鏈式調用來組合多個異步操作。
CompletableFuture 的特點
- 手動完成：可以手動設置
CompletableFuture的結果或異常。 - 鏈式調用：支持多個
CompletableFuture的鏈式調用，形成復雜的異步任務流。 - 組合操作：提供了豐富的方法來組合多個異步任務，例如
thenCombine、thenAcceptBoth等。 - 異常處理：提供了靈活的異常處理機制，可以在任務鏈中處理異常。
CompletableFuture 的底層原理
工作機制
CompletableFuture 的核心是基于 ForkJoinPool 實現的。ForkJoinPool 是一種特殊的線程池，適用于并行計算任務。它采用了工作竊取算法，能夠有效利用多核 CPU 的性能。
當我們提交一個任務給 CompletableFuture 時，它會將任務提交到默認的 ForkJoinPool.commonPool() 中執(zhí)行。我們也可以指定自定義的線程池來執(zhí)行任務。
狀態(tài)管理
CompletableFuture 具有以下幾種狀態(tài)：
- 未完成（Pending）：任務尚未完成。
- 完成（Completed）：任務已經成功完成，并返回結果。
- 異常（Exceptionally Completed）：任務在執(zhí)行過程中拋出了異常。
這些狀態(tài)通過內部的 volatile 變量來管理，并使用 CAS（Compare-And-Swap） 操作保證線程安全。
任務調度
CompletableFuture 的任務調度機制基于 ForkJoinPool 的工作竊取算法。當一個線程完成當前任務后，會從其他線程的任務隊列中竊取任務執(zhí)行，從而提高 CPU 利用率。
下面我們通過一個簡單的示例代碼來理解 CompletableFuture 的基本用法。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 創(chuàng)建一個 CompletableFuture 實例
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
return "Hello, World!";
});
// 阻塞等待結果
String result = future.get();
System.out.println(result);
}
}在上面的示例中，我們創(chuàng)建了一個 CompletableFuture 實例，并使用 supplyAsync 方法異步執(zhí)行任務。
supplyAsync 方法會將任務提交到默認的 ForkJoinPool 中執(zhí)行。最后，我們使用 get 方法阻塞等待結果并打印輸出。
鏈式調用
CompletableFuture 的一個重要特性是支持鏈式調用。
通過鏈式調用，我們可以將多個異步任務組合在一起，形成一個任務流。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureChainExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
return "Hello, World!";
}).thenApply(result -> {
return result + " from CompletableFuture";
}).thenApply(String::toUpperCase);
String finalResult = future.get();
System.out.println(finalResult);
}
}在這個示例中，我們使用 thenApply 方法對前一個任務的結果進行處理，并返回一個新的 CompletableFuture 實例。
通過鏈式調用，我們可以將多個任務串聯在一起，形成一個任務流。
組合操作
CompletableFuture 提供了多種方法來組合多個異步任務。以下是一些常用的組合操作示例：
1.thenCombine：組合兩個 CompletableFuture，并將兩個任務的結果進行處理。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureCombineExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);
CompletableFuture<Integer> combinedFuture = future1.thenCombine(future2, Integer::sum);
System.out.println(combinedFuture.get()); // 輸出 15
}
}2. thenAcceptBoth：組合兩個 CompletableFuture，并對兩個任務的結果進行消費處理。
import java.util.concurrent.CompletableFuture;
public class CompletableFutureAcceptBothExample {
public static void main(String[] args) {
CompletableFuture<Integer> future1 = CompletableFuture.supplyAsync(() -> 5);
CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> 10);
future1.thenAcceptBoth(future2, (result1, result2) -> {
System.out.println("Result: " + (result1 + result2));
}).join();
}
}3. allOf：組合多個 CompletableFuture，并在所有任務完成后執(zhí)行操作。
import java.util.concurrent.CompletableFuture;
public class CompletableFutureAllOfExample {
public static void main(String[] args) {
CompletableFuture<Void> future1 = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
System.out.println("Task 1 completed");
});
CompletableFuture<Void> future2 = CompletableFuture.runAsync(() -> {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
throw new IllegalStateException(e);
}
System.out.println("Task 2 completed");
});
CompletableFuture<Void> combinedFuture = CompletableFuture.allOf(future1, future2);
combinedFuture.join();
System.out.println("All tasks completed");
}
}異常處理
在異步任務中處理異常是非常重要的。CompletableFuture 提供了多種方法來處理任務執(zhí)行過程中的異常。
1.exceptionally：在任務拋出異常時，提供一個默認值。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureExceptionallyExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (true) {
throw new RuntimeException("Exception occurred");
}
return "Hello, World!";
}).exceptionally(ex -> {
System.out.println("Exception: " + ex.getMessage());
return "Default Value";
});
System.out.println(future.get()); // 輸出 Default Value
}
}2. handle：無論任務是否拋出異常，都進行處理。
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
public class CompletableFutureHandleExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
if (true) {
throw new RuntimeException("Exception occurred");
}
return "Hello, World!";
}).handle((result, ex) -> {
if (ex != null) {
return "Default Value";
}
return result;
});
System.out.println(future.get()); // 輸出 Default Value
}
}實戰(zhàn)案例：構建異步數據處理管道
為了更好地理解 CompletableFuture 的實際應用，我們來構建一個異步數據處理管道。
假設我們有一個數據源，需要對數據進行一系列的處理操作，并將處理結果輸出到文件中。
數據源模擬
我們首先模擬一個數據源，該數據源會生成一系列數據。
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
public class DataSource {
public List<Integer> getData() {
return IntStream.range(0, 10).boxed().collect(Collectors.toList());
}
}數據處理
接下來，我們定義數據處理操作。
假設我們需要對數據進行兩步處理：首先對每個數據乘以 2，然后對結果進行累加。
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.stream.Collectors;
public class DataProcessor {
public List<Integer> processStep1(List<Integer> data) {
return data.stream().map(x -> x * 2).collect(Collectors.toList());
}
public Integer processStep2(List<Integer> data) {
return data.stream().reduce(0, Integer::sum);
}
public CompletableFuture<List<Integer>> processStep1Async(List<Integer> data) {
return CompletableFuture.supplyAsync(() -> processStep1(data));
}
public CompletableFuture<Integer> processStep2Async(List<Integer> data) {
return CompletableFuture.supplyAsync(() -> processStep2(data));
}
}結果輸出
我們定義一個方法將處理結果輸出到文件中。
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.CompletableFuture;
public class ResultWriter {
public void writeResult(String fileName, Integer result) throws IOException {
Files.write(Paths.get(fileName), result.toString().getBytes());
}
public CompletableFuture<Void> writeResultAsync(String fileName, Integer result) {
return CompletableFuture.runAsync(() -> {
try {
writeResult(fileName, result);
} catch (IOException e) {
throw new IllegalStateException(e);
}
});
}
}主程序
最后，我們在主程序中將上述組件組合在一起，構建異步數據處理管道。
import java.util.List;
import java.util.concurrent.CompletableFuture;
public class Main {
public static void main(String[] args) {
DataSource dataSource = new DataSource();
DataProcessor dataProcessor = new DataProcessor();
ResultWriter resultWriter = new ResultWriter();
List<Integer> data = dataSource.getData();
CompletableFuture<List<Integer>> step1Future = dataProcessor.processStep1Async(data);
CompletableFuture<Integer> step2Future = step1Future.thenCompose(dataProcessor::processStep2Async);
CompletableFuture<Void> writeFuture = step2Future.thenCompose(result -> resultWriter.writeResultAsync("result.txt", result));
writeFuture.join();
System.out.println("Data processing completed");
}
}在這個例子中，我們使用 CompletableFuture 將數據處理步驟和結果輸出串聯在一起，形成了一個完整的異步數據處理管道。
通過 thenCompose 方法，我們將前一個任務的結果傳遞給下一個異步任務，從而實現了鏈式調用。
總結
本文深入探討了 CompletableFuture 的底層原理，展示了其工作機制，并通過多個代碼示例說明了如何在實際應用中使用 CompletableFuture。通過理解 CompletableFuture 的異步編程模型、狀態(tài)管理、任務調度和異常處理機制，我們可以更好地利用這一強大的工具構建高效、非阻塞的 Java 應用程序。
希望這篇文章能夠幫助你全面理解 CompletableFuture，并在實際開發(fā)中靈活應用。這些僅為個人經驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
Java虛擬機的調優(yōu)是一個復雜而關鍵的任務,可以通過多種參數來實現,本文就來介紹一下JVM調優(yōu)參數的設置,具有一定的參考價值,感興趣的可以了解一下2024-03-03
SpringBoot官網推薦使用HTML視圖解析器,但是根據個人的具體業(yè)務也有可能使用到JSP視圖解析器，所以本文介紹了兩種視圖解析器，感興趣的可以了解下2020-06-06
Maven包沖突導致NoSuchMethodError錯誤的解決辦法
web 項目 能正常編譯,運行時也正常啟動,但執(zhí)行到需要調用 org.codehaus.jackson 包中的某個方法時,產生運行異常,這篇文章主要介紹了Maven包沖突導致NoSuchMethodError錯誤的解決辦法,需要的朋友可以參考下2024-05-05
Java JDK動態(tài)代理（AOP）的實現原理與使用詳析
所謂代理，就是一個人或者一個機構代表另一個人或者另一個機構采取行動。下面這篇文章主要給大家介紹了關于Java JDK動態(tài)代理（AOP）實現原理與使用的相關資料，文中通過示例代碼介紹的非常詳細，需要的朋友可以參考借鑒，下面來一起看看吧。2017-07-07
這篇文章主要介紹了Spring的數據庫開發(fā)，主要圍繞SpringJDBC和Spring Jdbc Template兩個技術來講解，文中有詳細的代碼示例，需要的小伙伴可以參考一下2023-04-04
本文主要介紹了SpringSecurity 自定義認證登錄的項目實踐,以手機驗證碼登錄為例,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2024-08-08
java如何確定一個鏈表有環(huán)及入口節(jié)點
這篇文章主要介紹了java如何確定一個鏈表有環(huán)及入口節(jié)點，想了解數據結構的同學可以參考下2021-04-04
java比較器Comparable接口與Comaprator接口的深入分析
本篇文章是對java比較器Comparable接口與Comaprator接口進行了詳細的分析介紹，需要的朋友參考下2013-06-06
這篇文章主要介紹了SpringBoot中實現代理方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2023-06-06
簡單工廠模式是屬于創(chuàng)建型模式，又叫做靜態(tài)工廠方法（Static Factory Method）模式，但不屬于23種GOF設計模式之一。簡單工廠模式是由一個工廠對象決定創(chuàng)建出哪一種產品類的實例。簡單工廠模式是工廠模式家族中最簡單實用的模式，可以理解為是不同工廠模式的一個特殊實現2022-03-03