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

深入淺出多執行緒（三）：Future 非同步模式與 JDK 1.5 併發包實現全解析

編輯：Java 相關知識

前言：多執行緒協作的核心痛點

執行緒 A 需要執行緒 B 的計算結果，傳統做法是 join() 阻塞等待，嚴重浪費 CPU 資源。Future 模式 完美解決此問題：主執行緒提交任務後立即獲得「未來結果」佔位符，繼續處理其他業務，需要結果時再阻塞獲取，實現真正的非同步協作。

一、Future 模式核心概念

金融類比：期權。現在（9月）以約定價格購買未來（12月）的石油。若價格上漲，12月仍按 9月價格買入獲利。對應程式：

提交任務 = 買入期權（獲得 Future 物件）
執行業務邏輯 = 等待到期日
呼叫 get() = 行使期權（獲取真實結果）

經典應用：瀏覽器圖片載入——先顯示模糊縮圖（佔位符），下載完成後無縫替換高清圖。

二、JDK 1.5 前：手工實現 Future 模式

1. 結果持有類

public class FutureResult {
    private String result;
    private boolean isFinish = false;

    public synchronized String getResult() { return result; }
    public synchronized void setResult(String result) {
        this.result = result;
        this.isFinish = true;
    }
    public synchronized boolean isFinish() { return isFinish; }
}

2. 計算執行緒

public class GenerateResultThread extends Thread {
    private FutureResult futureResult;
    public GenerateResultThread(FutureResult fr) { this.futureResult = fr; }
    public void run() {
        try { Thread.sleep(5000); } catch (InterruptedException ignored) {}
        futureResult.setResult("ResultByGenerateResultThread");
    }
}

3. 主執行緒輪詢

FutureResult fr = new FutureResult();
new GenerateResultThread(fr).start();
Thread.sleep(2000);
if (!fr.isFinish()) System.out.println("還沒完成，繼續幹別的事");
Thread.sleep(4000);
if (fr.isFinish()) System.out.println("結果：" + fr.getResult());

缺點

需自行維護同步狀態、輪詢邏輯
無超時控制、無取消機制
程式碼冗長、易出錯

三、JDK 1.5 併發包：Callable + FutureTask 標準實現

核心介面與類

Callable<V>：V call() throws Exception 可回傳結果、可拋異常
Future<V>：get()、get(timeout)、cancel()、isDone()、isCancelled()
FutureTask<V>：實作 Runnable + Future，**雙重身份**可直接丟入 Thread 或 ExecutorService

標準用法

FutureTask<String> futureTask = new FutureTask<>(() -> {
    Thread.sleep(5000);
    return "ResultByGenerateResultThread";
});
new Thread(futureTask).start(); // 或 executor.submit(futureTask)

Thread.sleep(2000);
if (!futureTask.isDone()) System.out.println("還沒完成");

Thread.sleep(4000);
if (futureTask.isDone()) {
    try {
        System.out.println("結果：" + futureTask.get()); // 同步獲取，若未完成則阻塞
    } catch (ExecutionException e) {
        e.getCause().printStackTrace(); // 取得任務內部異常
    }
}

進階特性

超時獲取：futureTask.get(3, TimeUnit.SECONDS) 超時拋 TimeoutException
取消任務：futureTask.cancel(true) 嘗試中斷執行中任務
異常傳播：get() 會將任務異常包裝為 ExecutionException

四、FutureTask 內部原理深度解析

狀態機模型

NEW → COMPLETING → NORMAL / EXCEPTIONAL / CANCELLED
           ↓
       INTERRUPTING → INTERRUPTED

使用 volatile int state 保證可見性
CAS 原子更新狀態，無鎖高併發

關鍵方法實現

run() 執行任務並設定結果

public void run() {
    if (state != NEW || !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran = false;
            try {
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
                return;
            }
            if (ran) set(result);
        }
    } finally {
        runner = null;
    }
}

set(V result) 完成正常結果

protected void set(V v) {
    if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
        outcome = v;
        UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final 寫入
        finishCompletion(); // 喚醒所有等待執行緒
    }
}

get() 獲取結果（含阻塞邏輯）

public V get() throws InterruptedException, ExecutionException {
    int s = state;
    if (s <= COMPLETING) s = awaitDone(false, 0L);
    return report(s);
}
private int awaitDone(boolean timed, long nanos) {
    final long deadline = timed ? System.nanoTime() + nanos : 0L;
    Thread q = null;
    for (;;) {
        if (Thread.interrupted()) { removeWaiter(q); throw new InterruptedException(); }
        int s = state;
        if (s > COMPLETING) {
            if (q != null) q.thread = null;
            return s;
        }
        // 入隊等待...
        // 超時/中斷檢查...
        LockSupport.parkNanos(this, nanos); // 核心阻塞點
    }
}

五、ExecutorService 整合：生產環境標準用法

ExecutorService executor = Executors.newFixedThreadPool(10);
Future<String> future = executor.submit(() -> {
    // 業務邏輯
    return "result";
});

// 非阻塞檢查
if (future.isDone()) { ... }

// 阻塞獲取（含超時）
try {
    String result = future.get(5, TimeUnit.SECONDS);
} catch (TimeoutException e) {
    future.cancel(true); // 超時取消
} catch (ExecutionException e) {
    log.error("任務異常", e.getCause());
}

// 批次提交與併發控制
List<Future<String>> futures = executor.invokeAll(tasks); // 等待全部完成
String firstResult = executor.invokeAny(tasks); // 只要最快的一個

六、CompletableFuture（Java 8+）現代化升級

雖本文聚焦 JDK 1.5，但實戰強烈建議升級 CompletableFuture：

鏈式非阻塞回調：thenApply/thenAccept/thenCompose
組合式編排：allOf/anyOf
異常處理：exceptionally/handle
預設執行緒池：ForkJoinPool.commonPool()

結語

Future 模式是多執行緒協作的基石，JDK 1.5 透過 Callable/Future/FutureTask 提供了工業級標準實現，核心亮點在於：狀態機 + CAS + LockSupport 無鎖高效同步。掌握其內部機制，不僅能正確使用 API（避免 get() 無限阻塞、異常吞沒等坑），更能在設計非同步框架時借鑑其精髓。現代開發請優先擁抱 CompletableFuture 或 Reactive Streams，享受聲明式非同步程式設計的紅利。