在 Java 中完成線程的作業後返回一個值
一、概述
Java 的主要特性之一是並發性。它允許多個線程運行並執行並行任務。因此，我們可以執行異步和非阻塞指令。這將優化可用資源，尤其是當計算機具有多個 CPU 時。有兩種類型的線程：有返回值或沒有（在後一種情況下，我們說它將有一個 void 返回方法）。
在本文中，我們將重點關注如何從其作業已終止的線程返回一個值。
Thread和Runnable
我們將 Java 線程稱為輕量級進程。讓我們看一下 Java 程序通常是如何工作的：
Java 程序是一個正在執行的過程。線程是 Java 進程的子集，可以訪問主內存。它可以與同一進程的其他線程通信。
一個線程有一個生命週期和不同的狀態。實現它的一種常見方法是通過Runnable接口：
public class RunnableExample implements Runnable {
...
@Override
public void run() {
// do something
}
}然後，我們可以開始我們的線程：
Thread thread = new Thread(new RunnableExample());
thread.start();
thread.join();如我們所見，我們無法從Runnable返回值。但是，我們可以使用wait()和notify()與其他線程同步。 join()方法將使執行處於等待狀態，直到它完成。當我們從異步執行中獲得結果時，我們稍後會看到這有多重要。
3. Callable
Java 從 1.5 版本開始引入了Callable接口。讓我們看一個異步任務返回階乘計算值的示例。我們使用BigInteger作為結果可以是一個很大的數字：
public class CallableFactorialTask implements Callable<BigInteger> {
// fields and constructor
@Override
public BigInteger call() throws Exception {
return factorial(BigInteger.valueOf(value));
}
}讓我們也創建一個簡單的階乘計算器：
public class FactorialCalculator {
public static BigInteger factorial(BigInteger end) {
BigInteger start = BigInteger.ONE;
BigInteger res = BigInteger.ONE;
for (int i = start.add(BigInteger.ONE).intValue(); i <= end.intValue(); i++) {
res = res.multiply(BigInteger.valueOf(i));
}
return res;
}
public static BigInteger factorial(BigInteger start, BigInteger end) {
BigInteger res = start;
for (int i = start.add(BigInteger.ONE).intValue(); i <= end.intValue(); i++) {
res = res.multiply(BigInteger.valueOf(i));
}
return res;
}
}Callable只有一個方法call()我們需要覆蓋。該方法將返回我們的異步任務的對象。
Callable和Runnable都是@FunctionalInterface 。 Callable可以返回一個值並拋出異常。但是，它需要一個Future來完成任務。
4.執行Callable
我們可以使用Future或 Fork/Join 執行Callable 。
4.1. Future Callable
從 1.5 版本開始，Java 有了Future接口來創建包含我們異步處理的響應的對象。我們可以在邏輯上將Future與 Javascript 中的Promise進行比較。
例如，當我們想要從多個端點獲取數據時，我們通常會看到Future 。因此，我們需要等待所有任務完成才能收集響應數據。
Future包裝響應並等待線程完成。但是，我們可能會遇到中斷，例如超時或執行異常。
讓我們看一下Future接口：
public interface Future<V> {
boolean cancel(boolean var1);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long var1, TimeUnit var3) throws InterruptedException, ExecutionException, TimeoutException;
}get方法很有趣，我們可以等待並獲取執行的結果。
要啟動Future作業，我們將其執行與ThreadPool相關聯。這樣，我們將為那些異步任務分配一些資源。
讓我們用一個Executor創建一個示例，該 Executor 對我們之前看到的Callable中的階乘數求和。我們將使用Executor接口和ExecutorService實現來創建ThreadPoolExecutor 。我們可能想使用固定的或緩存的線程池。在這種情況下，我們將使用緩存線程池來演示：
public BigInteger execute(List<CallableFactorialTask> tasks) {
BigInteger result = BigInteger.ZERO;
ExecutorService cachedPool = Executors.newCachedThreadPool();
List<Future<BigInteger>> futures;
try {
futures = cachedPool.invokeAll(tasks);
} catch (InterruptedException e) {
// exception handling example
throw new RuntimeException(e);
}
for (Future<BigInteger> future : futures) {
try {
result = result.add(future.get());
} catch (InterruptedException | ExecutionException e) {
// exception handling example
throw new RuntimeException(e);
}
}
return result;
}我們可以用一個圖表來表示這個執行，我們可以在其中觀察線程池和Callable是如何交互的：
Executor將在Future對像中調用和收集所有內容。然後，我們可以從我們的異步處理中得到一個或多個結果。
讓我們通過總結兩個階乘數的結果來測試：
@Test
void givenCallableExecutor_whenExecuteFactorial_thenResultOk() {
BigInteger result = callableExecutor.execute(Arrays.asList(new CallableFactorialTask(5), new CallableFactorialTask(3)));
assertEquals(BigInteger.valueOf(126), result);
}4.2.可通過 Fork/Join Callable
我們還可以選擇使用ForkJoinPool 。它仍然與ExecutorSerivce類似，因為它擴展了AbstractExecutorService類。但是，它有不同的方式來創建和組織線程。它將任務分解為更小的任務並優化資源，使它們永遠不會閒置。我們可以用圖表表示子任務：
我們可以看到主任務將分叉為 SubTask1、SubTask3 和 SubTask4 作為最小的可執行文件。最後，他們將加入最終結果。
讓我們使用ForkJoinPool將前面的示例轉換為一個示例。我們可以將所有內容包裝在一個執行程序方法中：
public BigInteger execute(List<Callable<BigInteger>> forkFactorials) {
List<Future<BigInteger>> futures = forkJoinPool.invokeAll(forkFactorials);
BigInteger result = BigInteger.ZERO;
for (Future<BigInteger> future : futures) {
try {
result = result.add(future.get());
} catch (InterruptedException | ExecutionException e) {
// exception handling example
throw new RuntimeException(e);
}
}
return result;
}在這種情況下，我們只需要創建一個不同的池來獲得我們的期貨。讓我們用階乘Callable的列表來測試它：
@Test
void givenForkExecutor_whenExecuteCallable_thenResultOk() {
assertEquals(BigInteger.valueOf(126),
forkExecutor.execute(Arrays.asList(new CallableFactorialTask(5), new CallableFactorialTask(3))));
}然而，我們也可以決定如何分叉我們的任務。我們可能希望根據某些標準來分叉我們的計算，例如，根據輸入參數或服務負載。
我們需要將任務重寫為ForkJoinTask ，因此我們將使用RecursiveTask ：
public class ForkFactorialTask extends RecursiveTask<BigInteger> {
// fields and constructor
@Override
protected BigInteger compute() {
BigInteger factorial = BigInteger.ONE;
if (end - start > threshold) {
int middle = (end + start) / 2;
return factorial.multiply(new ForkFactorialTask(start, middle, threshold).fork()
.join()
.multiply(new ForkFactorialTask(middle + 1, end, threshold).fork()
.join()));
}
return factorial.multiply(factorial(BigInteger.valueOf(start), BigInteger.valueOf(end)));
}
}如果某個閾值適用，我們將細分我們的主要任務。然後我們可以使用invoke()方法來獲取結果：
ForkJoinPool forkJoinPool = ForkJoinPool.commonPool();
int result = forkJoinPool.invoke(forkFactorialTask);此外， submit()或execute()也是一個選項。但是，我們總是需要join()命令來完成執行。
我們還創建一個測試，我們在其中執行階乘任務：
@Test
void givenForkExecutor_whenExecuteRecursiveTask_thenResultOk() {
assertEquals(BigInteger.valueOf(3628800), forkExecutor.execute(new ForkFactorialTask(10, 5)));
}在這種情況下，我們會將 10 的階乘分成兩個任務。第一個將從 1 計算到 5，而第二個將從 6 計算到 10。
5. 可完成的未來
自 1.8 版以來，Java 通過引入**CompletableFuture**改進了多線程。它從Future執行中刪除了樣板代碼，並添加了鏈接或組合異步結果等功能。然而，最重要的是，我們現在可以對任何方法進行異步計算，因此我們不受Callable約束。此外，我們可以將語義不同的多個Futures連接在一起。
5.1. supplyAsync()
使用CompletableFuture可以很簡單：
CompletableFuture<BigInteger> future = CompletableFuture.supplyAsync(() -> factorial(BigInteger.valueOf(10)));
...
BigInteger result = future.get();我們不再需要Callable 。我們可以將任何 lambda 表達式作為參數傳遞。讓我們用supplyAsync()測試階乘方法：
@Test
void givenCompletableFuture_whenSupplyAsyncFactorial_thenResultOk() throws ExecutionException, InterruptedException {
CompletableFuture<BigInteger> completableFuture = CompletableFuture.supplyAsync(() -> factorial(BigInteger.valueOf(10)));
assertEquals(BigInteger.valueOf(3628800), completableFuture.get());
}請注意，我們沒有指定任何線程池。在這種情況下，將使用默認的ForkJoinPool 。但是，我們可以指定一個Executor ，例如，具有固定線程池：
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> factorial(BigInteger.valueOf(10)), Executors.newFixedThreadPool(1));5.2. thenCompose()
我們還可以創建一個連續的Futures鏈。假設我們有兩個階乘任務，第二個需要第一個的輸入：
CompletableFuture<BigInteger> completableFuture = CompletableFuture.supplyAsync(() -> factorial(BigInteger.valueOf(3)))
.thenCompose(inputFromFirstTask -> CompletableFuture.supplyAsync(() -> factorial(inputFromFirstTask)));
BigInteger result = completableFuture.get();我們可以使用thenCompose()方法來使用鏈中下一個CompletableFuture的返回值。
讓我們結合兩個階乘的執行。例如，我們從 3 開始，給我們一個階乘 6 作為下一個階乘的輸入：
@Test
void givenCompletableFuture_whenComposeTasks_thenResultOk() throws ExecutionException, InterruptedException {
CompletableFuture<BigInteger> completableFuture = CompletableFuture.supplyAsync(() -> factorial(BigInteger.valueOf(3)))
.thenCompose(inputFromFirstTask -> CompletableFuture.supplyAsync(() -> factorial(inputFromFirstTask)));
assertEquals(BigInteger.valueOf(720), completableFuture.get());
}5.3. allOf()
有趣的是，我們可以使用接受輸入 var-arg 的靜態方法allOf()並行執行多個Futures 。
從多個執行中收集異步結果就像添加到allOf()和join()來完成任務一樣簡單：
BigInteger result = allOf(asyncTask1, asyncTask2)
.thenApplyAsync(fn -> factorial(factorialTask1.join()).add(factorial(new BigInteger(factorialTask2.join()))), Executors.newFixedThreadPool(1)).join();請注意， allOf()的返回類型為 void 。因此，我們需要手動從單個Futures中獲取結果。此外，我們可以在同一執行中運行具有不同返回類型的Futures 。
為了進行測試，讓我們加入兩個不同的階乘任務。為了演示，一個有一個數字輸入，而第二個有一個字符串：
@Test
void givenCompletableFuture_whenAllOfTasks_thenResultOk() {
CompletableFuture<BigInteger> asyncTask1 = CompletableFuture.supplyAsync(() -> BigInteger.valueOf(5));
CompletableFuture<String> asyncTask2 = CompletableFuture.supplyAsync(() -> "3");
BigInteger result = allOf(asyncTask1, asyncTask2)
.thenApplyAsync(fn -> factorial(asyncTask1.join()).add(factorial(new BigInteger(asyncTask2.join()))), Executors.newFixedThreadPool(1))
.join();
assertEquals(BigInteger.valueOf(126), result);
}六，結論
在本教程中，我們了解瞭如何從線程返回對象。我們看到瞭如何將Callable與Future和線程池結合使用。 Future包裝結果並等待所有任務完成。我們還看到了ForkJoinPool的一個例子，可以將我們的執行優化為多個子任務。
Java 8 中的CompletableFuture工作方式類似，但還提供了新功能，例如可以執行任何 lambda 表達式。它還允許我們鏈接和組合異步任務的結果。
最後，我們用Future 、 Fork和CompletableFuture測試了一個簡單的階乘任務。
與往常一樣，我們可以在 GitHub 上找到工作代碼示例。