跳到主要内容

JEP 499: 结构化并发（第四预览版）

QWen Max 中英对照 JEP 499: Structured Concurrency (Fourth Preview)

概要​

通过引入结构化并发的 API 来简化并发编程。结构化并发将运行在不同线程中的一组相关任务视为一个工作单元，从而简化错误处理和取消操作，提高可靠性，并增强可观察性。这是一个预览 API。

历史​

Structured Concurrency 由 JEP 428 提出，并在 JDK 19 中作为孵化 API 发布。它通过 JEP 437 在 JDK 20 中再次孵化，做了一些小的更新以继承作用域值 (JEP 429)。它在 JDK 21 中通过 JEP 453 首次预览，其中将 StructuredTaskScope::fork(...) 更改为返回一个 Subtask，而不是一个 Future。它通过 JEP 462 在 JDK 22 中再次预览，并且通过 JEP 480 在 JDK 23 中再次预览，没有变化。

我们在此提议在 JDK 24 中再次预览该 API，不做任何更改，以便有更多时间收集实际使用中的反馈。

目标​

  • 推广一种并发编程风格，可以消除由取消和关闭引起的常见风险，例如线程泄漏和取消延迟。

  • 提高并发代码的可观察性。

非目标​

  • 目标不是要替换 java.util.concurrent 包中的任何并发结构，例如 ExecutorService 和 Future。

  • 目标不是为 Java 平台定义一个确定的结构化并发 API。其他结构化并发结构可以由第三方库定义，或在未来的 JDK 版本中定义。

  • 目标不是定义一种在线程之间共享数据流的方法（即 channels）。我们可能会在未来提出这样的建议。

  • 目标不是用新的线程取消机制替换现有的线程中断机制。我们可能会在未来提出这样的建议。

动机​

开发人员通过将任务分解为多个子任务来管理程序中的复杂性。在普通的单线程代码中，子任务是按顺序执行的。然而，如果子任务之间足够独立，并且有足够的硬件资源，那么可以通过并行执行子任务来使整个任务运行得更快（即，具有更低的延迟）。例如，一个由多个 I/O 操作结果组成任务，如果每个 I/O 操作都在自己的线程中并发执行，则会运行得更快。虚拟线程（JEP 444）使得为每个这样的 I/O 操作分配一个线程变得经济高效，但管理由此产生的大量线程仍然是一个挑战。

使用 ExecutorService 的非结构化并发​

java.util.concurrent.ExecutorService API 在 Java 5 中引入，帮助开发人员并发执行子任务。

例如，这里有一个方法 handle()，它表示服务器应用程序中的一个任务。它通过将两个子任务提交给 ExecutorService 来处理传入的请求。一个子任务执行方法 findUser()，另一个子任务执行方法 fetchOrder()。ExecutorService 立即为每个子任务返回一个 Future，并根据 Executor 的调度策略并发执行这些子任务。handle() 方法通过对其 future 的 get() 方法进行阻塞调用来等待子任务的结果，因此该任务被称为 join 了它的子任务。

Response handle() throws ExecutionException, InterruptedException {
    Future<String>  user  = esvc.submit(() -> findUser());
    Future<Integer> order = esvc.submit(() -> fetchOrder());
    String theUser  = user.get();   // Join findUser
    int    theOrder = order.get();  // Join fetchOrder
    return new Response(theUser, theOrder);
}

由于子任务是并发执行的，因此每个子任务可以独立成功或失败。（在这种情况下，失败意味着抛出异常。）通常，如果任何子任务失败，则像 handle() 这样的任务也应该失败。当发生失败时，理解线程的生命周期可能会出奇地复杂：

  • 如果 findUser() 抛出异常，那么在调用 user.get() 时 handle() 将会抛出异常，但 fetchOrder() 将会继续在其自己的线程中运行。这是一个线程泄漏，最理想的情况下是浪费资源；最坏的情况下，fetchOrder() 线程将会干扰其他任务。

  • 如果执行 handle() 的线程被中断，中断将不会传播到子任务。findUser() 和 fetchOrder() 线程都将泄漏，即使在 handle() 失败后仍将继续运行。

  • 如果 findUser() 执行时间很长，但在此期间 fetchOrder() 失败了，那么 handle() 会在 user.get() 上阻塞，而不是取消它，从而不必要地等待 findUser() 完成。只有在 findUser() 完成并且 user.get() 返回之后，order.get() 才会抛出异常，导致 handle() 失败。

在每种情况下，问题在于我们的程序在逻辑上是按照任务-子任务的关系结构化的，但这些关系只存在于开发者的脑海中。

这不仅会增加出错的空间，还会使诊断和解决此类错误变得更加困难。例如，可观察性工具如线程转储会在无关线程的调用堆栈上显示 handle()、findUser() 和 fetchOrder()，而不会提示任务-子任务的关系。

我们可能会尝试通过在发生错误时显式取消其他子任务来改进，例如通过使用 try-finally 包装任务，并在失败任务的 catch 块中调用其他任务的 future 的 cancel(boolean) 方法。我们还需要在 try-with-resources 语句 中使用 ExecutorService，如 JEP 444 中的示例所示，因为 Future 不提供等待已取消任务的方法。但所有这些都很难正确实现，并且通常会使代码的逻辑意图更难以理解。跟踪任务之间的关系，并手动添加所需的相互取消依赖关系，这对开发者来说要求过高。

需要手动协调生命周期的原因是 ExecutorService 和 Future 允许无限制的并发模式。参与的任何线程之间都没有约束或顺序。一个线程可以创建 ExecutorService，第二个线程可以向其提交工作，而执行工作的线程与第一个或第二个线程都没有关系。此外，在一个线程提交工作后，一个完全不同的线程可以等待执行的结果。任何拥有 Future 引用的代码都可以加入它，即通过调用 get() 等待其结果——甚至是在获取 Future 的线程之外的其他线程中的代码。实际上，由一个任务启动的子任务不必返回到提交它的任务。它可以返回到许多任务中的任何一个——或者可能不返回到任何任务。

由于 ExecutorService 和 Future 允许这样的非结构化使用，它们不会强制执行甚至跟踪任务和子任务之间的关系，即使这些关系是常见且有用的。因此，即使子任务是在同一个任务中提交和连接的，一个子任务的失败也不能自动导致另一个子任务的取消：在上述 handle() 方法中，fetchOrder() 的失败不能自动导致 findUser() 的取消。fetchOrder() 的 future 与 findUser() 的 future 无关，两者也与最终通过其 get() 方法连接它的线程无关。我们希望可靠地自动化这种取消操作，而不是要求开发人员手动管理。

任务结构应反映代码结构​

与 ExecutorService 下自由组合的线程不同，单线程代码的执行总是强制任务和子任务的层次结构。方法的主体块 {...} 对应一个任务，而在该块中调用的方法则对应子任务。被调用的方法必须返回或抛出异常给调用它的方法。被调用的方法不能比调用它的方法存活更久，也不能返回或抛给另一个不同的方法。因此，所有的子任务都在任务完成之前结束，每个子任务都是其父任务的子任务，并且每个子任务相对于其他子任务和任务的生命周期由代码的语法块结构决定。

例如，在这个单线程版本的 handle() 中，任务-子任务关系从语法结构上就可以看出来：

Response handle() throws IOException {
    String theUser  = findUser();
    int    theOrder = fetchOrder();
    return new Response(theUser, theOrder);
}

我们不会在 findUser() 子任务完成之前开始 fetchOrder() 子任务，无论 findUser() 是否成功。如果 findUser() 失败，那么我们根本不会开始 fetchOrder()，并且 handle() 任务会隐式失败。子任务只能返回其父任务这一事实是重要的：这意味着父任务可以将一个子任务的失败视为取消其他未完成子任务并使自身失败的触发器。

在单线程代码中，任务-子任务的层次结构在运行时通过调用栈具体化。因此，我们自然而然地得到了这种决定错误传播方式的父子关系。当观察单个线程时，这种层次关系是显而易见的：findUser()（以及后来的fetchOrder()）看起来是从属于handle()的。这使得回答“handle()现在正在处理什么？”这个问题变得很容易。

如果任务及其子任务之间的父子关系能够从代码的语法结构中清晰地体现出来，并且在运行时也能具体化，那么并发编程将会变得更简单、更可靠、更易于观察——就像单线程代码一样。语法结构将界定子任务的生命周期，并能够在运行时表示线程间的层次结构，类似于线程内的调用栈。这种表示方法可以实现错误传播和取消操作，同时也可以对并发程序进行有意义的观察。

（Java 平台已经有一个用于对并发任务施加结构的 API，即 java.util.concurrent.ForkJoinPool，它是并行流背后的执行引擎。但是，该 API 是为计算密集型任务设计的，而不是为涉及 I/O 的任务设计的。）

结构化并发​

结构化并发是一种并发编程方法，它保留了任务和子任务之间的自然关系，从而导致更易读、可维护和可靠的并发代码。术语“结构化并发”是由 Martin Sústrik 创造的，并由 Nathaniel J. Smith 推广开来。其他语言的思想，例如 Erlang 的分层监督者，在结构化并发的错误处理设计中有所体现。

结构化并发源于一个简单的原则，即

如果一个任务分解成并发的子任务，那么它们都会返回到同一个地方，即该任务的代码块。

在结构化并发中，子任务代表一个任务工作。该任务等待子任务的结果并监控它们是否失败。与单线程代码的结构化编程技术一样，多线程的结构化并发的力量来源于两个思想：代码块中执行流的明确定义的入口和出口点，以及操作生命周期的严格嵌套，这种方式反映了它们在代码中的语法嵌套。

由于代码块的进入和退出点定义明确，因此并发子任务的生命周期被限制在其父任务的语法块内。由于兄弟子任务的生命周期嵌套在它们的父任务的生命周期内，因此可以将它们作为一个整体进行推理和管理。由于父任务的生命周期又嵌套在其父任务的生命周期内，运行时可以将任务层次结构实体化为一棵树，这是单个线程的调用栈的并发对应物。这允许代码对整个任务子树应用策略（如截止时间），并允许可观察性工具将子任务呈现为其父任务的从属任务。

结构化并发与虚拟线程非常匹配，虚拟线程是由 JDK 实现的轻量级线程。许多虚拟线程共享同一个操作系统线程，从而允许存在大量的虚拟线程。除了数量众多之外，虚拟线程还足够廉价，可以表示任何并发行为单元，甚至是涉及 I/O 的行为。这意味着服务器应用程序可以使用结构化并发一次处理数千或数百万个传入请求：它可以为处理每个请求的任务分配一个新的虚拟线程，当任务通过提交子任务进行并发执行而分散时，它可以为每个子任务分配一个新的虚拟线程。在幕后，通过安排每个虚拟线程携带对其唯一父线程的引用，将任务-子任务关系具体化为一棵树，类似于调用堆栈中的帧如何引用其唯一的调用者。

总之，虚拟线程提供了大量的线程。结构化并发可以正确且稳健地协调这些线程，并使可观测性工具能够以开发人员理解的方式显示线程。在 Java 平台上拥有结构化并发的 API 将使构建可维护、可靠和可观察的服务器应用程序变得更加容易。

描述​

结构化并发 API 的主要类是 java.util.concurrent 包中的 StructuredTaskScope。此类允许开发人员将任务构建为一组并行子任务，并将其作为一个单元进行协调。子任务通过单独 fork 它们并在之后作为一个单元 join 它们，以及可能作为一个单元取消它们，在其自己的线程中执行。子任务的成功结果或异常由父任务聚合和处理。StructuredTaskScope 将子任务的生命周期限制在一个明确的词法作用域内，在该作用域内，任务与其子任务的所有交互——包括 fork、join、取消、错误处理和结果组合——都发生在这个范围内。

这里是前面的 handle() 示例，改写为使用 StructuredTaskScope（ShutdownOnFailure 的解释见下面）：

Response handle() throws ExecutionException, InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        Supplier<String>  user  = scope.fork(() -> findUser());
        Supplier<Integer> order = scope.fork(() -> fetchOrder());

        scope.join()            // Join both subtasks
             .throwIfFailed();  // ... and propagate errors

        // Here, both subtasks have succeeded, so compose their results
        return new Response(user.get(), order.get());
    }
}

与原始示例相比，理解这里涉及的线程的生命周期很容易：在所有条件下，它们的生命周期都限制在一个词法范围内，即 try-with-resources 语句的主体。此外，使用 StructuredTaskScope 确保了许多有价值的特性：

  • 错误处理与短路 — 如果 findUser() 或 fetchOrder() 子任务中的任何一个抛出异常而失败，另一个尚未完成的任务将被取消。（这是由 ShutdownOnFailure 实现的关闭策略管理的；其他策略也是可能的）。

  • 取消传播 — 如果运行 handle() 的线程在调用 join() 之前或期间被中断，当线程退出作用域时，两个子任务都会自动被取消。

  • 清晰性 — 上述代码具有清晰的结构：设置子任务，等待它们完成或被取消，然后决定是成功（并处理已完成的子任务的结果）还是失败（子任务已经完成，因此无需进一步清理）。

  • 可观测性 — 如下所述可观测性中描述的线程转储清楚地显示了任务层次结构，执行 findUser() 和 fetchOrder() 的线程作为作用域的子级显示。

StructuredTaskScope 是一个预览 API，默认情况下是禁用的​

要使用 StructuredTaskScope API，您必须启用预览 API，如下所示：

  • 使用 javac --release 24 --enable-preview Main.java 编译程序，并使用 java --enable-preview Main 运行它；或者，

  • 当使用 source code launcher 时，使用 java --enable-preview Main.java 运行程序；或者，

  • 当使用 jshell 时，使用 jshell --enable-preview 启动它。

使用 StructuredTaskScope​

public class StructuredTaskScope<T> implements AutoCloseable {

    public <U extends T> Subtask<U> fork(Callable<? extends U> task);
    public void shutdown();

    public StructuredTaskScope<T> join() throws InterruptedException;
    public StructuredTaskScope<T> joinUntil(Instant deadline)
        throws InterruptedException, TimeoutException;
    public void close();

    protected void handleComplete(Subtask<? extends T> handle);
    protected final void ensureOwnerAndJoined();

}

使用 StructuredTaskScope 的代码的一般工作流程是：

  1. 创建一个作用域。创建该作用域的线程是它的所有者。

  2. 使用 fork(Callable) 方法在作用域中派生子任务。

  3. 在任何时候，任何子任务或作用域的所有者都可以调用作用域的 shutdown() 方法来取消未完成的子任务，并防止派生新的子任务。

  4. 作用域的所有者作为一个整体加入该作用域，即其所有子任务。所有者可以调用作用域的 join() 方法，等待所有子任务完成（成功或失败）或通过 shutdown() 取消。或者，它可以调用作用域的 joinUntil(java.time.Instant) 方法，等待到一个截止时间。

  5. 加入后，处理子任务中的任何错误并处理它们的结果。

  6. 关闭作用域，通常通过 try-with-resources 隐式完成。这会关闭作用域（如果尚未关闭），并等待任何已取消但尚未完成的子任务完成。

每次调用 fork(...) 都会启动一个新的线程来执行子任务，默认情况下这是一个虚拟线程。子任务可以创建自己的嵌套 StructuredTaskScope 来派生自己的子任务，从而创建一个层次结构。该层次结构反映在代码的块结构中，这限制了子任务的生命周期：一旦作用域关闭，所有子任务的线程都保证已经终止，并且在块退出时不会留下任何线程。

在任何作用域中的子任务、嵌套作用域中的任何子子任务以及作用域的所有者都可以随时调用该作用域的 shutdown() 方法，以表示任务已完成 — 即使其他子任务仍在执行。shutdown() 方法中断 仍在执行子任务的线程，并导致 join() 或 joinUntil(Instant) 方法返回。因此，所有子任务都应该以响应中断的方式编写。在调用 shutdown() 后分叉的新子任务将处于 UNAVAILABLE 状态，并且不会运行。实际上，shutdown() 是顺序代码中 break 语句的并发类比。

在作用域内调用 join() 或 joinUntil(Instant) 是强制性的。如果在加入之前作用域的块就退出了，那么作用域将等待所有子任务终止，然后抛出一个异常。

作用域的所有者线程可能在 join 之前或期间被中断。例如，它可能是已经被关闭的外部作用域的子任务。如果发生这种情况，join() 和 joinUntil(Instant) 将抛出异常，因为继续执行已经没有意义。然后 try-with-resources 语句将关闭作用域，这会取消所有子任务并等待它们终止。这样做的效果是自动将任务的取消传播到其子任务。如果 joinUntil(Instant) 方法的截止时间在子任务终止或调用 shutdown() 之前到期，则它将抛出异常，并且 try-with-resources 语句将再次关闭作用域。

当 join() 成功完成时，每个子任务要么已成功完成，要么已失败，要么因为在范围关闭时被取消。

一旦加入，作用域的所有者将处理失败的子任务，并处理成功完成的子任务的结果；这通常由关闭策略处理（见下文）。可以通过 Subtask.get() 方法获取成功完成的任务的结果。get() 方法永远不会阻塞；如果在加入之前或子任务未成功完成时误调用该方法，则会抛出 IllegalStateException。

在某个范围内分叉的子任务会继承 ScopedValue 绑定（JEP 446）。如果范围的所有者从绑定的 ScopedValue 中读取值，那么每个子任务将读取相同的值。

如果一个作用域的所有者本身是现有作用域的子任务，也就是说，它是在派生子任务时创建的，那么该作用域就成为新作用域的父作用域。因此，作用域和子任务形成了一个树形结构。

StructuredTaskScope 的结构化使用是在运行时强制执行的。例如，尝试从不在该作用域树层次结构中的线程（即，所有者、子任务以及嵌套作用域中的子任务（子-子任务））调用 fork(Callable) 将会抛出异常。在 try-with-resources 块之外使用作用域并在没有调用 close() 的情况下返回，或者没有保持 close() 调用的正确嵌套，可能会导致作用域的方法抛出 StructureViolationException。

StructuredTaskScope 对并发操作强制实施结构和顺序。因此，它没有实现 ExecutorService 或 Executor 接口，因为这些接口的实例通常以非结构化的方式使用（见下文）。然而，将使用 ExecutorService 的代码迁移到 StructuredTaskScope 是很简单的，特别是那些能够从结构中受益的代码。

实际上，大多数 StructuredTaskScope 的使用都不会直接利用 StructuredTaskScope 类，而是使用下一节中描述的实现关闭策略的两个子类之一。在其他场景中，用户可能会编写自己的子类来实现自定义的关闭策略。

关机策略​

在处理并发子任务时，通常会使用短路模式来避免不必要的工作。有时这样做是有意义的，例如，如果其中一个子任务失败（即，调用所有）或成功（即，调用任一），则取消所有子任务。StructuredTaskScope 的两个子类，ShutdownOnFailure 和 ShutdownOnSuccess，通过策略支持这些模式，当第一个子任务失败或成功时，分别关闭作用域。

关闭策略还提供了集中处理异常的方法，可能还包括成功的结果。这符合结构化并发的精神，即整个作用域被视为一个单元。

这里是一个具有失败时关闭策略的 StructuredTaskScope（也在上面的 handle() 示例中使用），它可以并发运行一组任务，如果其中任何一个任务失败，则整个任务失败：

<T> List<T> runAll(List<Callable<T>> tasks) 
        throws InterruptedException, ExecutionException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
        List<? extends Supplier<T>> suppliers = tasks.stream().map(scope::fork).toList();
        scope.join()
             .throwIfFailed();  // Propagate exception if any subtask fails
        // Here, all tasks have succeeded, so compose their results
        return suppliers.stream().map(Supplier::get).toList();
    }
}

这里是一个具有成功时关闭策略的 StructuredTaskScope，它返回第一个成功子任务的结果：

<T> T race(List<Callable<T>> tasks, Instant deadline) 
        throws InterruptedException, ExecutionException, TimeoutException {
    try (var scope = new StructuredTaskScope.ShutdownOnSuccess<T>()) {
        for (var task : tasks) {
            scope.fork(task);
        }
        return scope.joinUntil(deadline)
                    .result();  // Throws if none of the subtasks completed successfully
    }
}

一旦某个子任务成功，此作用域将自动关闭，取消未完成的子任务。如果所有子任务都失败或给定的截止时间已过，则该任务失败。这种模式在例如需要从一组冗余服务中的任何一个获取结果的服务器应用程序中可能很有用。

虽然提供了这两种关闭策略，但开发人员可以创建抽象其他模式的自定义策略（见下文）。

处理结果​

通常，作用域所有者唯一会调用的 Subtask 方法是 get() 方法。其他所有的 Subtask 方法通常只会在自定义关闭策略的 handleComplete(...) 方法实现中使用（见下文自定义关闭策略）。事实上，我们建议将 fork(...) 返回的 Subtask 引用变量类型声明为，例如 Supplier<String> 而不是 Subtask<String>（当然，如果你选择使用 var 则另当别论）。如果关闭策略本身处理子任务的结果——就像 ShutdownOnSuccess 的情况一样——那么应该完全避免使用 fork(...) 返回的 Subtask 对象，并将 fork(...) 方法视为返回 void。子任务应将其结果返回任何信息，这些信息应在策略集中处理异常后由作用域所有者处理。

如果作用域所有者处理子任务异常以生成复合结果，而不是使用关闭策略，则可以将异常作为值从子任务中返回。例如，以下方法并行运行任务列表，并返回一个已完成的 Future 列表，其中包含每个任务的成功或异常结果：

<T> List<Future<T>> executeAll(List<Callable<T>> tasks)
        throws InterruptedException {
    try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    	  List<? extends Supplier<Future<T>>> futures = tasks.stream()
    	      .map(task -> asFuture(task))
     	      .map(scope::fork)
     	      .toList();
    	  scope.join();
    	  return futures.stream().map(Supplier::get).toList();
    }
}

static <T> Callable<Future<T>> asFuture(Callable<T> task) {
   return () -> {
       try {
           return CompletableFuture.completedFuture(task.call());
       } catch (Exception ex) {
           return CompletableFuture.failedFuture(ex);
       }
   };
}

自定义关闭策略​

StructuredTaskScope 可以被扩展，并且可以覆盖其受保护的 handleComplete(...) 方法，以实现不同于 ShutdownOnSuccess 和 ShutdownOnFailure 的策略。例如，子类可以

  • 收集成功完成的子任务的结果，并忽略失败的子任务，
  • 在子任务失败时收集异常，或者
  • 调用 shutdown() 方法来关闭并导致 join() 在某些条件出现时唤醒。

当子任务完成时，即使在调用 shutdown() 之后，它也会作为 Subtask 报告给 handleComplete(...) 方法：

public sealed interface Subtask<T> extends Supplier<T> {
    enum State { SUCCESS, FAILED, UNAVAILABLE }

    State state();
    Callable<? extends T> task();
    T get();
    Throwable exception();
}

handleComplete(...) 方法会在子任务完成时被调用，无论其是成功完成（SUCCESS 状态）还是失败完成（FAILED 状态），前提是 shutdown() 尚未被调用。只有当子任务处于 SUCCESS 状态时才能调用 get() 方法，而只有当子任务处于 FAILED 状态时才能调用 exception() 方法；在其他情况下调用 get() 或 exception() 会导致抛出 IllegalStateException。UNAVAILABLE 状态表示以下情况之一：(1) 子任务已被分叉但尚未完成；(2) 子任务在关闭后完成；或 (3) 子任务在关闭后被分叉，因此尚未开始。对于处于 UNAVAILABLE 状态的子任务，永远不会调用 handleComplete(...) 方法。

子类通常会定义一些方法，以便在 join() 方法返回后执行的代码能够获取结果、状态或其他结果。一个收集结果并忽略失败子任务的子类可以定义一个返回结果集合的方法。一个实现在子任务失败时关闭策略的子类可以定义一个获取第一个失败子任务的异常的方法。

以下是一个 StructuredTaskScope 子类的示例，它收集成功完成的子任务的结果。它定义了 results() 方法，供主任务用来检索结果。

class MyScope<T> extends StructuredTaskScope<T> {

    private final Queue<T> results = new ConcurrentLinkedQueue<>();

    MyScope() { super(null, Thread.ofVirtual().factory()); }

    @Override
    protected void handleComplete(Subtask<? extends T> subtask) {
        if (subtask.state() == Subtask.State.SUCCESS)
            results.add(subtask.get());
    }

    @Override
    public MyScope<T> join() throws InterruptedException {
        super.join();
        return this;
    }

    // Returns a stream of results from the subtasks that completed successfully
    public Stream<T> results() {
        super.ensureOwnerAndJoined();
        return results.stream();
    }

}

此自定义策略可以这样使用：

<T> List<T> allSuccessful(List<Callable<T>> tasks) throws InterruptedException {
    try (var scope = new MyScope<T>()) {
        for (var task : tasks) scope.fork(task);
        return scope.join()
                    .results().toList();
    }
}

合流场景​

上面的示例集中在扇出场景，即管理多个并发的出站 I/O 操作。StructuredTaskScope 在扇入场景中也很有用，这种场景下需要管理多个并发的入站 I/O 操作。在这种场景中，我们通常会根据入站请求创建数量未知的子任务。

以下是一个在 StructuredTaskScope 内派生子任务以处理传入连接的服务器示例：

void serve(ServerSocket serverSocket) throws IOException, InterruptedException {
    try (var scope = new StructuredTaskScope<Void>()) {
        try {
            while (true) {
                var socket = serverSocket.accept();
                scope.fork(() -> handle(socket));
            }
        } finally {
            // If there's been an error or we're interrupted, we stop accepting
            scope.shutdown();  // Close all active connections
            scope.join();
        }
    }
}

从并发的角度来看，这种场景的不同之处不在于请求的方向，而在于任务的持续时间和数量。在这里，与前面的例子不同，作用域所有者的持续时间是无界的——只有在被中断时它才会停止。子任务的数量也是未知的，因为它们是根据外部事件动态派生的。

所有处理连接的子任务都在该范围内创建，因此在线程转储中很容易看出它们的用途，线程转储会将它们显示为范围所有者的子级。作为一个整体关闭整个服务也很容易。

可观察性​

我们扩展了由 JEP 444 添加的新 JSON 线程转储格式，以显示 StructuredTaskScope 将线程分组到层次结构中的情况：

$ jcmd <pid> Thread.dump_to_file -format=json <file>

每个作用域的 JSON 对象包含在该作用域中派生的线程数组及其堆栈跟踪。作用域的所有者线程通常会在 join 方法中被阻塞，等待子任务完成；线程转储通过显示结构化并发所施加的树形层次结构，使查看子任务线程的行为变得容易。作用域的 JSON 对象还包含对其父对象的引用，以便可以从转储中重新构建程序的结构。

com.sun.management.HotSpotDiagnosticsMXBean API 也可以用于生成这样的线程转储，可以直接使用，也可以通过平台 MBeanServer 和本地或远程 JMX 工具间接使用。

为什么 fork(...) 不返回一个 Future？​

当 StructuredTaskScope API 处于孵化阶段时，fork(...) 方法返回一个 Future。这通过使 fork(...) 类似于现有的 ExecutorService::submit 方法，提供了一种熟悉感。然而，鉴于 StructuredTaskScope 的使用方式与 ExecutorService 不同——以一种结构化的方式使用，如上所述——使用 Future 带来的困惑多于清晰。

  • 熟悉的 Future 用法涉及调用其 get() 方法，该方法会阻塞直到结果可用。但在 StructuredTaskScope 的上下文中，以这种方式使用 Future 不仅不被鼓励，而且是适得其反的。结构化的 Future 对象只应在 join() 返回后查询，在那时它们已知已完成或被取消，并且应该使用的方法不是熟悉的 get()，而是新引入的 resultNow()，它永远不会阻塞。

  • 一些开发者想知道为什么 fork(...) 没有返回功能更强大的 CompletableFuture 对象。由于由 fork(...) 返回的 Future 只应在已知完成之后才使用，因此 CompletableFuture 并没有提供任何好处，因为它的高级功能仅对未完成的 future 有用。此外，CompletableFuture 是为异步编程范式设计的，而 StructuredTaskScope 则鼓励阻塞范式。

    简而言之，Future 和 CompletableFuture 被设计为提供了在结构化并发中适得其反的自由度。

  • 结构化并发是指将运行在不同线程中的多个任务视为一个工作单元，而 Future 在将多个任务视为单个任务时最有用。作用域应仅阻塞一次以等待其子任务的结果，然后应集中处理异常。因此，在绝大多数情况下，唯一应该在从 fork(...) 返回的 Future 上调用的方法是 resultNow()。这与通常使用 Future 的方式有很大的不同，而 Future 接口在这种情况下分散了对其正确使用的注意力。

在当前的 API 中，Subtask::get() 的行为与 API 孵化时 Future::resultNow() 的行为完全相同。

替代方案​

  • 增强 ExecutorService 接口。我们为该接口实现了一个原型，该原型始终强制执行结构，并限制哪些线程可以提交任务。然而，我们发现这样做存在问题，因为 JDK 和生态系统中大多数对 ExecutorService（及其父接口 Executor）的使用都是非结构化的。将相同的 API 用于一个更加受限的概念很可能会引起混淆。例如，在大多数情况下，将一个结构化的 ExecutorService 实例传递给接受该类型的现有方法几乎肯定会抛出异常。