您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 相關知識

Java 多線程 Future 與 Callable 實戰：異步計算結果聚合模式

編輯：Java 相關知識

核心痛點

多線程並行計算時，主線程需要等待所有子線程完成並收集結果。傳統 Thread.join() + 共享變數方式繁瑣且線程不安全。Callable + Future 提供標準化異步計算模型。

關鍵介面對比

介面	方法	返回值	拋出異常	用途
Runnable	run()	void	不可拋受檢異常	無返回值任務
Callable<V>	call()	V	可拋任意異常	有返回值任務
Future<V>	get()/get(timeout)	V	InterruptedException, ExecutionException	異步結果句柄

標準模式：提交任務 → 獲取 Future → 阻塞收集結果

ExecutorService executor = Executors.newFixedThreadPool(10);
List> futures = new ArrayList<>();

// 1. 提交 100 個 Callable 任務
for (int i = 0; i < 100; i++) {
    final int taskId = i;
    Future future = executor.submit(new Callable() {
        @Override public Integer call() throws Exception {
            // 模擬耗時計算
            Thread.sleep(ThreadLocalRandom.current().nextInt(100));
            return taskId * 2; // 返回計算結果
        }
    });
    futures.add(future);
}

// 2. 主線程收集所有結果（按提交順序阻塞）
List results = new ArrayList<>();
for (Future future : futures) {
    try {
        results.add(future.get()); // 阻塞直到完成
    } catch (InterruptedException | ExecutionException e) {
        log.error("任務失敗", e);
    }
}

// 3. 聚合處理
int sum = results.stream().mapToInt(Integer::intValue).sum();
System.out.println("總和: " + sum);

executor.shutdown();

進階模式：CompletableFuture 組合式異步

Java 8+ 推薦使用 CompletableFuture，支援非阻塞鏈式組合。

並行聚合：allOf / anyOf

List> cfs = IntStream.range(0, 100)
    .mapToObj(i -> CompletableFuture.supplyAsync(() -> {
        sleep(random(100));
        return i * 2;
    }, executor))
    .collect(toList());

// 等待全部完成，無需手動 get()
CompletableFuture all = CompletableFuture.allOf(cfs.toArray(new CompletableFuture[0]));
all.thenRun(() -> {
    List results = cfs.stream()
        .map(CompletableFuture::join) // 已完成，非阻塞
        .collect(toList());
    int sum = results.stream().mapToInt(Integer::intValue).sum();
    System.out.println("異步聚合總和: " + sum);
}).join(); // 主線程等待聚合完成

流水線組合：thenApply / thenCombine

CompletableFuture f1 = CompletableFuture.supplyAsync(() -> heavyCalc1(), executor);
CompletableFuture f2 = CompletableFuture.supplyAsync(() -> heavyCalc2(), executor);

// 兩個並行計算完成後，合併結果
CompletableFuture combined = f1.thenCombine(f2, (r1, r2) -> r1 + r2);

combined.thenAccept(sum -> System.out.println("組合結果: " + sum));

異常處理：exceptionally / handle

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    if (random() > 0.5) throw new RuntimeException("失敗");
    return "成功";
}, executor).exceptionally(ex -> {
    log.warn("降級處理", ex);
    return "默認值";
});

ExecutorService 選型指南

工廠方法	適用場景	特點
newFixedThreadPool(n)	固定並發數，長期運行任務	隊列無界，可能 OOM
newCachedThreadPool()	大量短生命任務	線程可回收，無界創建，風險大
newSingleThreadExecutor()	串行化執行	保證順序
newWorkStealingPool()	CPU 密集型、分治任務	ForkJoinPool，工作窺探
自定義 ThreadPoolExecutor	生產環境	有界隊列、拒絕策略、監控指標

生產級線程池配置範本

@Configuration
public class ThreadPoolConfig {
    @Bean("taskExecutor")
    public ThreadPoolExecutor taskExecutor() {
        return new ThreadPoolExecutor(
            10, 50,                    // core / max
            60L, TimeUnit.SECONDS,     // keepAlive
            new LinkedBlockingQueue<>(1000), // 有界隊列
            new ThreadFactoryBuilder().setNameFormat("biz-worker-%d").build(),
            new ThreadPoolExecutor.CallerRunsPolicy() // 拒絕策略：調用者執行
        );
    }
}

最佳實踐總結

優先用 CompletableFuture，代替手動 Future.get() 循環，代碼更聲明式、組合性更強。
異步邊界清晰：Controller 層返回 CompletableFuture，Spring MVC 自動異步處理；Service 層內部組合。
線程池隔離：不同業務類型（IO 密集/CPU 密集）使用獨立線程池，避免相互擠佔。
必須處理異常：future.get() 拋出 ExecutionException 包裝真實異常，CompletableFuture.exceptionally() 定義降級邏輯。
關閉釋放：應用關閉時調用 executor.shutdown() + awaitTermination()，防止任務丟失。