解讀CompletableFuture的底層原理
引言
在現(xiàn)代 Java 編程中，異步編程變得越來越重要。為了實現(xiàn)高效和非阻塞的代碼，Java 8 引入了 CompletableFuture，一個用于構建異步應用程序的強大工具。
本文將詳細探討 CompletableFuture 的底層原理，展示其工作機制，并通過代碼示例說明如何在實際應用中使用它。
異步編程的背景
異步編程是指在程序運行過程中，不等待某個操作完成，而是繼續(xù)執(zhí)行其他操作，待異步操作完成后再處理其結果。這樣可以提高程序的效率，特別是在 I/O 操作和網(wǎng)絡請求等耗時操作中。
在 Java 8 之前，實現(xiàn)異步編程主要依賴于 Future 接口。然而，Future 存在一些局限性，例如無法手動完成、不能鏈式調用等。為了解決這些問題，Java 8 引入了 CompletableFuture。
什么是 CompletableFuture
CompletableFuture 是 Java 8 中新增的類，實現(xiàn)了 Future 和 CompletionStage 接口，提供了強大的異步編程能力。
CompletableFuture 允許以非阻塞的方式執(zhí)行任務，并且可以通過鏈式調用來組合多個異步操作。
CompletableFuture 的特點
- 手動完成：可以手動設置
CompletableFuture的結果或異常。 - 鏈式調用：支持多個
CompletableFuture的鏈式調用，形成復雜的異步任務流。 - 組合操作：提供了豐富的方法來組合多個異步任務，例如
thenCombine、thenAcceptBoth等。 - 異常處理：提供了靈活的異常處理機制，可以在任務鏈中處理異常。
CompletableFuture 的底層原理
工作機制
CompletableFuture 的核心是基于 ForkJoinPool 實現(xiàn)的。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 實例。
通過鏈式調用，我們可以將多個任務串聯(lián)在一起，形成一個任務流。
組合操作
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)案例：構建異步數(shù)據(jù)處理管道
為了更好地理解 CompletableFuture 的實際應用，我們來構建一個異步數(shù)據(jù)處理管道。
假設我們有一個數(shù)據(jù)源，需要對數(shù)據(jù)進行一系列的處理操作，并將處理結果輸出到文件中。
數(shù)據(jù)源模擬
我們首先模擬一個數(shù)據(jù)源，該數(shù)據(jù)源會生成一系列數(shù)據(jù)。
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());
}
}數(shù)據(jù)處理
接下來，我們定義數(shù)據(jù)處理操作。
假設我們需要對數(shù)據(jù)進行兩步處理：首先對每個數(shù)據(jù)乘以 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);
}
});
}
}主程序
最后，我們在主程序中將上述組件組合在一起，構建異步數(shù)據(jù)處理管道。
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 將數(shù)據(jù)處理步驟和結果輸出串聯(lián)在一起，形成了一個完整的異步數(shù)據(jù)處理管道。
通過 thenCompose 方法，我們將前一個任務的結果傳遞給下一個異步任務，從而實現(xiàn)了鏈式調用。
總結
本文深入探討了 CompletableFuture 的底層原理，展示了其工作機制，并通過多個代碼示例說明了如何在實際應用中使用 CompletableFuture。通過理解 CompletableFuture 的異步編程模型、狀態(tài)管理、任務調度和異常處理機制，我們可以更好地利用這一強大的工具構建高效、非阻塞的 Java 應用程序。
希望這篇文章能夠幫助你全面理解 CompletableFuture，并在實際開發(fā)中靈活應用。這些僅為個人經驗，希望能給大家一個參考，也希望大家多多支持腳本之家。
相關文章
Spring MVC實現(xiàn)mysql數(shù)據(jù)庫增刪改查完整實例
這篇文章主要介紹了Spring MVC實現(xiàn)mysql數(shù)據(jù)庫增刪改查完整實例，從創(chuàng)建一個web項目開始，分享了項目結構以及具體Java代碼和前端頁面等相關內容，具有一定借鑒價值，需要的朋友可以了解下。2017-12-12
在當今數(shù)字化時代,微服務架構憑借其高可擴展性、靈活性和易于維護等優(yōu)勢,成為了眾多企業(yè)構建大型應用系統(tǒng)的首選架構模式,當我們將微服務部署在 Linux 服務器上時,有時會遭遇令人頭疼的死鎖問題,本位給大家介紹了如何排查定位Java中的死鎖,需要的朋友可以參考下2025-02-02
人之初性本善，但是隨著自身的經歷、生活環(huán)境等因素的影響，人逐漸會生出貪嗔癡。實際上不光人有貪念，我們的算法也會有貪念，今天就和大家介紹下一個有貪念的算法模型---貪心算法，看看一個算法是怎么產生貪念的2022-05-05
文章介紹了JVM的多個配置參數(shù),如堆內存設置、線程棧大小、年輕代與年老代的配置等,其中包括標準參數(shù)和各種非標準參數(shù)（-X和-XX）,并提到了它們對JVM性能和垃圾回收機制的影響,還區(qū)分了Client和Server模式2025-10-10
在JavaScript中實現(xiàn)異步編程的方法多種多樣,每種方法都有其特定的應用場景和優(yōu)缺點,這篇文章主要介紹了JavaScript異步通信三種玩法的相關資料,文中通過代碼介紹的非常詳細,需要的朋友可以參考下2025-12-12
這篇文章主要介紹了Java之Algorithm_analysis案例詳解,本篇文章通過簡要的案例,講解了該項技術的了解與使用,以下就是詳細內容,需要的朋友可以參考下2021-09-09
本篇文章主要介紹了springMVC的生命周期詳解，小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-03-03
這篇文章主要介紹了Java Dubbo框架知識點梳理,通過詳細的文字講解和代碼實例,梳理了Dubbo這個框架,需要的朋友可以參考下2021-06-06
Spring動態(tài)管理定時任務之ThreadPoolTaskScheduler解讀
這篇文章主要介紹了Spring動態(tài)管理定時任務之ThreadPoolTaskScheduler解讀，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-12-12
Cache在ConcurrentHashMap的基礎上提供了自動加載數(shù)據(jù)、清除數(shù)據(jù)、get-if-absend-compute的功能,本文給大家介紹如何使用Guava Cache做緩存,感興趣的朋友一起看看吧2023-11-11