【Java学習｜実務向け】Javaの非同期処理を極める！CompletableFutureで並行処理をスマートに

2. 処理の連鎖 (Chain):
非同期処理が完了した後に、その結果を受け取って次の処理を実行します。

• `thenApply(Function fn)`: 前の処理の結果を使って新しい値を生成します。
• `thenAccept(Consumer action)`: 前の処理の結果を受け取って、何かを実行します（値を返さない）。
• `thenRun(Runnable action)`: 前の処理が完了したことをトリガーに、何かを実行します（結果は使わない、値を返さない）。

3. 合成 (Combine):
複数の`CompletableFuture`を組み合わせて、より複雑な非同期処理を構築します。

• `thenCombine(CompletionStage other, BiFunction fn)`: 2つの`CompletableFuture`が両方完了したときに、それぞれの結果を使って新しい値を生成します。
• `allOf(CompletableFuture… cfs)`: 指定した全ての`CompletableFuture`が完了するのを待ちます。
• `anyOf(CompletableFuture… cfs)`: 指定した`CompletableFuture`のうち、いずれか1つが完了するのを待ちます。

4. エラーハンドリング:
非同期処理中に発生した例外を捕捉し、適切に処理します。

• `exceptionally(Function fn)`: 例外が発生した場合に、代替値を返します。
• `handle(BiFunction fn)`: 正常終了時も例外発生時も、結果または例外を受け取って処理します。

4. サンプルプログラム：実践的なコード例

ここでは、複数の非同期処理を実行し、その結果を組み合わせて最終的な結果を得る例を示します。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CompletableFutureExample {

public static void main(String[] args) {
// スレッドプールを準備します。
// 複数の非同期タスクを実行するためにExecutorServiceを使用します。
// ここではCPUコア数に応じた固定スレッドプールを作成します。
ExecutorService executor = Executors.newFixedThreadPool(4);

System.out.println(“非同期処理を開始します…”);

// 最初の非同期処理: ユーザーIDを取得する
CompletableFuture userIdFuture = CompletableFuture.supplyAsync(() -> {
System.out.println(“ユーザーID取得中…”);
try {
// ネットワーク通信やDBアクセスを模倣
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(“ユーザーID取得中にエラー”, e);
}
System.out.println(“ユーザーID取得完了。”);
return “user123”;
}, executor); // 実行に使うExecutorServiceを指定

// 2番目の非同期処理: ユーザーIDからユーザー情報を取得する
CompletableFuture userInfoFuture = userIdFuture.thenApplyAsync((userId) -> {
// userIdFutureの結果（userId）を受け取る
System.out.println(“ユーザー情報取得中 (ID: ” + userId + “)…”);
try {
// ネットワーク通信やDBアクセスを模倣
TimeUnit.SECONDS.sleep(3);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(“ユーザー情報取得中にエラー”, e);
}
System.out.println(“ユーザー情報取得完了。”);
return “Name: Alice, Email: alice@example.com”;
}, executor); // thenApplyAsyncでもExecutorServiceを指定可能

// 3番目の非同期処理: ユーザーIDから注文履歴を取得する
CompletableFuture orderHistoryFuture = userIdFuture.thenApplyAsync((userId) -> {
// userIdFutureの結果（userId）を受け取る
System.out.println(“注文履歴取得中 (ID: ” + userId + “)…”);
try {
// ネットワーク通信やDBアクセスを模倣
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException(“注文履歴取得中にエラー”, e);
}
System.out.println(“注文履歴取得完了。”);
return “Order1: ItemA, Order2: ItemB”;
}, executor);

// 2つの処理（ユーザー情報と注文履歴）が両方完了したら、それらを結合する
CompletableFuture combinedResultFuture = userInfoFuture.thenCombineAsync(orderHistoryFuture, (userInfo, orderHistory) -> {
// userInfoFutureとorderHistoryFutureの両方の結果を受け取る
System.out.println(“結果を結合中…”);
return “User Info: ” + userInfo + “\nOrder History: ” + orderHistory;
}, executor);

// 最終的な結果を取得する（ブロッキングします）
try {
System.out.println(“最終結果を待機中…”);
String finalResult = combinedResultFuture.get(); // ここで完了を待つ
System.out.println(“\n— 最終結果 —“);
System.out.println(finalResult);
System.out.println(“—————-“);
} catch (InterruptedException | ExecutionException e) {
System.err.println(“処理中に例外が発生しました: ” + e.getMessage());
e.printStackTrace();
} finally {
// ExecutorServiceをシャットダウンします。
// これを行わないと、プログラムが終了しない可能性があります。
System.out.println(“ExecutorServiceをシャットダウンします。”);
executor.shutdown();
}

System.out.println(“メインスレッド処理終了。”);
}
}

このサンプルでは、まず`supplyAsync`で非同期タスクを開始し、その結果を`thenApplyAsync`で次のタスクに渡しています。さらに、`thenCombineAsync`を使って複数の非同期タスクの結果を結合しています。`get()`メソッドは、すべての非同期処理が完了するまでメインスレッドをブロックしますが、これは最終結果を取得するために必要です。

5. 応用・注意点：現場で役立つTips

• デフォルトのExecutor: `supplyAsync`や`thenApplyAsync`などのメソッドで`ExecutorService`を指定しない場合、`ForkJoinPool.commonPool()`がデフォルトで使用されます。しかし、I/Oバウンドな処理が多い場合、この共通プールではスレッドが枯渇しやすくパフォーマンスが低下する可能性があります。そのため、I/Oバウンドな処理には専用の`ExecutorService`（`Executors.newCachedThreadPool()`や`Executors.newFixedThreadPool()`など）を用意し、明示的に指定することをお勧めします。
• Virtual Threads (Project Loom): Java 21で正式リリースされたVirtual Threadsは、軽量なスレッドであり、I/Oバウンドな処理との相性が抜群です。CompletableFutureとVirtual Threadsを組み合わせることで、より少ないリソースで高い並行性を実現できる可能性があります。例えば、`Executors.newVirtualThreadPerTaskExecutor()`のようなExecutorServiceを使うことで、CompletableFutureのタスクがVirtual Threads上で実行されるようになります。
• Structured Concurrency: Javaの将来的なバージョンでは、Structured Concurrencyの概念が導入される可能性があります。これは、関連する非同期タスクをグループ化し、そのグループ全体のライフサイクルを管理することで、エラーハンドリングやリソース管理をより安全かつ容易にするためのアプローチです。CompletableFutureはその基盤技術として重要になります。
• タイムアウト処理: `get(long timeout, TimeUnit unit)`メソッドを使用すると、指定した時間内に完了しない場合に`TimeoutException`をスローさせることができます。これにより、無限に待機することを防ぎます。
• エラーハンドリングの重要性: 非同期処理では、例外がどこで発生したかを追跡するのが難しくなりがちです。`exceptionally`や`handle`を適切に使い、例外発生時のフォールバック処理やログ記録を実装することが、堅牢なアプリケーションを構築する上で非常に重要です。

CompletableFutureを使いこなすことで、Javaでの非同期処理が格段に書きやすくなります。ぜひ、日々の開発に取り入れてみてください。