Structured Concurrency

1. 簡介

當執行緒各自為政

一般的多執行緒常常像沒有指揮的排練。每個執行緒就像不聽彼此的樂手，各自演奏自己的旋律。有人提早結束去抽菸，有人卡在同一個和弦，有人甚至彈錯音而丟出錯誤。結果不是交響樂，而是刺耳的雜音：很難弄清楚誰在哪裡出錯，更別說要一次把大家都停下來——這根本是場硬仗。

Structured Concurrency 解決的正是這個問題。它把零散的執行緒變成真正的合奏：所有任務都在同一位「指揮」之下。只要指揮下令停止——整個樂團就安靜。如果有一位樂手搞砸了——其他人會被妥善停止，而不會破壞整體和諧。所有結果與錯誤都集中收集，而不是散落在程式碼各處。

想像一下：你不再讓樂手各自亂演，而是把他們集合到同一個演奏廳。有指揮、有譜面，即使小號走音——樂團也不會崩盤，而是體面地結束演出。

Structured Concurrency 帶來什麼

• 單一「scope」的任務：所有子任務都活在同一段程式碼區塊中，它們的生命週期受此區塊限制。
• 可預測的結束：父執行緒在所有子任務結束前不會結束。
• 集中式取消：只要某個任務失敗，或父方決定結束——所有子任務都會被正確取消。
• 一致性的錯誤處理：子任務的錯誤會被彙整，你可以取得「原因樹」（tree of causes）。
• 乾淨且可讀的程式碼：沒有「懸掛」的執行緒，沒有被遺忘的任務，也沒有競相取消的混亂。

Structured Concurrency 不只是新的 API，而是一種新的思維模式：任務應當像一般的程式碼區塊一樣被結構化（例如 try-with-resources）。

2. Java 中 Structured Concurrency 的狀態

在撰寫本課程時，Structured Concurrency 的狀態是 Preview（Java 21–23），但預期會在 Java 24/25 推進到 GA（General Availability）。API 位於套件 jdk.incubator.concurrent。在投入生產前，請務必查閱你所用 JDK 版本的最新發行說明！

主要類別：

• StructuredTaskScope —— 用於管理一組任務的基礎類別。
• 變體：StructuredTaskScope.ShutdownOnFailure、StructuredTaskScope.ShutdownOnSuccess —— 任務的結束策略。

StructuredTaskScope 的核心概念

模型：fork、join 與結果點名

當指揮（也就是父任務）發出信號——子任務就各自去演奏自己的段落。這個時刻稱為 fork——就像把樂手分派到不同的廳房演奏各自的片段。

接著就會進入 join —— 指揮舉起指揮棒，大家回到台上，一起奏出最後的和弦。

然後你可以逐一詢問每位參與者，表演如何：

• 透過 resultNow() 立即取得結果（若一切順利且無錯誤）；
• 透過 throwIfFailed() —— 確認沒有人走音。若有人真的亂了譜，會丟出統一的例外——就像指揮說：「我們的樂團出了狀況，重新來過」。

結束策略

每位指揮都有自己的停樂規則。在 Structured Concurrency 中，這由 結束策略 來決定：

• ShutdownOnFailure —— 只要有任何樂手失了拍，指揮揮手：「停！重來」。其他人立刻停止演奏。
• ShutdownOnSuccess —— 相反地，只要有人完美演完自己的段落，指揮就滿意地說：「夠了，不用再繼續，我們已經有贏家了」。其他人會安靜——也就是「第一個成功的答案」策略。

使用虛擬執行緒

每個子任務都由 StructuredTaskScope 在虛擬執行緒中啟動。這就像擁有一個懂事又迅速、不挑場地也不挑剔的樂團。你可以放心地建立數百、數千個這樣的執行者——它們不是笨重的系統執行緒，而是幾乎無重量的音符，只在需要時發聲。

3. 範例：HTTP 請求聚合器

來看一個實務情境：我們有三個資料來源（例如三個不同的伺服器），我們希望要嘛拿到全部的回應並加以聚合，要嘛取得第一個成功回應者的結果。

方案一：「全部都必須成功」（ShutdownOnFailure）

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

public class AggregatorAllSuccess {
    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Future<String> f1 = scope.fork(() -> fetchFromSource1());
            Future<String> f2 = scope.fork(() -> fetchFromSource2());
            Future<String> f3 = scope.fork(() -> fetchFromSource3());

            scope.join(); // 等待所有任務完成
            scope.throwIfFailed(); // 只要有任何一個失敗，就拋出例外

            // 所有任務都成功——可以聚合結果
            String result = f1.resultNow() + f2.resultNow() + f3.resultNow();
            System.out.println("聚合結果: " + result);
        }
    }

    static String fetchFromSource1() { /* ... */ return "A"; }
    static String fetchFromSource2() { /* ... */ return "B"; }
    static String fetchFromSource3() { /* ... */ return "C"; }
}

發生了什麼：

• 三個任務同時啟動（於虛擬執行緒中）。
• 只要有任何一個失敗——其他任務就會被取消，並拋出例外。
• 若全部成功——即可安全地聚合結果。

方案二：「第一個有效成功」（ShutdownOnSuccess）

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

public class AggregatorFirstSuccess {
    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
            Future<String> f1 = scope.fork(() -> fetchFromSource1());
            Future<String> f2 = scope.fork(() -> fetchFromSource2());
            Future<String> f3 = scope.fork(() -> fetchFromSource3());

            scope.join(); // 等待第一個成功
            scope.throwIfFailed(); // 若全部失敗——就拋出例外

            String result = scope.result(); // 第一個成功任務的結果
            System.out.println("第一個成功的結果: " + result);
        }
    }

    static String fetchFromSource1() { /* ... */ return "A"; }
    static String fetchFromSource2() { /* ... */ return "B"; }
    static String fetchFromSource3() { /* ... */ return "C"; }
}

發生了什麼：

• 一旦有任務成功完成——其餘任務就會被取消。
• 如果全部失敗——會拋出例外。

4. 自動取消與降級

StructuredTaskScope 會在策略需要時，自動負責取消其餘任務。舉例來說，若有任務失敗（ShutdownOnFailure），或有任務成功完成（ShutdownOnSuccess），其餘任務會收到取消信號（interrupt）。

範例：帶有逾時的妥善結束

import jdk.incubator.concurrent.StructuredTaskScope;
import java.time.Instant;
import java.util.concurrent.Future;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> f1 = scope.fork(() -> fetchWithTimeout());
    Future<String> f2 = scope.fork(() -> fetchWithTimeout());

    scope.joinUntil(Instant.now().plusSeconds(2)); // 最多等待 2 秒
    scope.throwIfFailed();

    String result = f1.resultNow() + f2.resultNow();
    System.out.println(result);
}

若任務未在 2 秒內完成——會拋出例外，所有任務都會被取消。

5. 錯誤與例外處理

子任務拋出的例外如何被傳遞到 scope

有時演出中還是會有人彈錯音——StructuredTaskScope 不會裝作什麼都沒發生。它會仔細記錄是誰出了差錯，然後把完整報告交給指揮。當你呼叫 throwIfFailed() 時，它會丟出彙整過的例外——有點像總結報告：「以下是今天走音的名單」。必要時，你可以展開這棵「原因樹」，看看究竟是誰拖了後腿。而如果你只想知道某位演出者的情況——Future.exceptionNow() 會告訴你他的段落是如何收場的。

何時取消不等於失敗

務必記得：取消任務不代表錯誤。如果指揮宣布「演出結束」，樂手只是收好樂器——這是 cancelled，而不是 failed。只有真正演錯的情況才算錯誤，這種例外才會被彙入總結。

範例：原因樹

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> f1 = scope.fork(() -> { throw new RuntimeException("錯誤 1"); });
    Future<String> f2 = scope.fork(() -> { throw new RuntimeException("錯誤 2"); });

    scope.join();
    scope.throwIfFailed(); // 會拋出包含兩個原因的例外
} catch (Exception e) {
    e.printStackTrace();
    // 可透過 e.getSuppressed() 取得 suppressed exceptions
}

6. 與 CompletableFuture 的比較

StructuredTaskScope 與 CompletableFuture 都能啟動平行任務，但：

• StructuredTaskScope 適合任務在邏輯上相關、且應一起結束/取消的情況（任務階層）。
• CompletableFuture 適合沒有階層關係的任務組合（例如轉換鏈、反應式情境）。

何時 StructuredTaskScope 能簡化程式碼：

• 需要保證離開區塊前，所有子任務都已完成。
• 需要集中式取消與錯誤處理。
• 必須確保不會留下「懸掛」的任務。

何時 CompletableFuture 更方便：

• 任務彼此無關，可以獨立存續。
• 需要複雜的組合（thenCombine、thenCompose 等）。

7. 實作：HTTP 請求聚合器

任務：向 3 個來源發送請求，取得第一個成功的回應

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

public class HttpAggregator {
    public static void main(String[] args) throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
            Future<String> f1 = scope.fork(() -> httpRequest("https://api1.example.com"));
            Future<String> f2 = scope.fork(() -> httpRequest("https://api2.example.com"));
            Future<String> f3 = scope.fork(() -> httpRequest("https://api3.example.com"));

            scope.join();
            scope.throwIfFailed();

            String result = scope.result();
            System.out.println("第一個成功的回應: " + result);
        }
    }

    static String httpRequest(String url) throws Exception {
        // 模擬請求（可使用 HttpClient）
        Thread.sleep((long) (Math.random() * 1000));
        if (Math.random() < 0.3) throw new RuntimeException("請求錯誤: " + url);
        return "來自 " + url + " 的回應";
    }
}

任務：若有任務失敗——正確地關停其餘任務

import jdk.incubator.concurrent.StructuredTaskScope;
import java.util.concurrent.Future;

try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    Future<String> f1 = scope.fork(() -> httpRequest("https://api1.example.com"));
    Future<String> f2 = scope.fork(() -> httpRequest("https://api2.example.com"));

    scope.join();
    scope.throwIfFailed();

    String result = f1.resultNow() + f2.resultNow();
    System.out.println("兩個回應: " + result);
} catch (Exception e) {
    System.err.println("其中一個任務出現錯誤: " + e.getMessage());
}

8. 使用 StructuredTaskScope 的常見錯誤

錯誤 #1：忘了呼叫 join() 或 throwIfFailed()。
若不呼叫 join()，任務可能在離開區塊前尚未完成。若不呼叫 throwIfFailed()，子任務的錯誤會被忽略。

錯誤 #2：在任務完成前嘗試取得結果。
在任務完成前呼叫 resultNow() 會丟出 IllegalStateException。請先透過 join() 等候完成。

錯誤 #3：忽略取消。
如果任務已被取消（例如因為 scope 的策略），請不要嘗試取得它的結果——否則會拋出例外。

錯誤 #4：混用不同的結束策略。
不應在 scope 內手動取消任務——請使用 ShutdownOnFailure 或 ShutdownOnSuccess 的策略。

錯誤 #5：在虛擬執行緒中執行耗時的 CPU-bound 任務。
StructuredTaskScope 預設使用虛擬執行緒——它們非常適合 I/O-bound 任務，但不會加速繁重的計算。

錯誤 #6：忘了關閉 scope（未使用 try-with-resources）。
StructuredTaskScope 實作了 AutoCloseable —— 請務必使用 try-with-resources，以保證所有任務都會被妥善結束。