JEP 437：结构化并发（第二次孵化）
总结
通过引入 结构化并发 的 API 来简化多线程编程。结构化并发将不同线程中运行的多个任务视为一个单一的工作单元，从而简化错误处理和取消操作，提高可靠性，并增强可观测性。这是一个 孵化中的 API。
历史
重新孵化的 API 唯一的变化是 StructuredTaskScope 得到了更新，以支持在任务范围内创建的线程继承作用域值（JEP 429）。这简化了跨线程共享不可变数据的过程。
目标
-
提高多线程代码的可维护性、可靠性和可观测性。
-
推广一种并发编程风格，这种风格可以消除因取消和关闭操作而引发的常见风险，例如线程泄漏和取消延迟。
非目标
-
并不旨在替换
java.util.concurrent包中的任何并发结构，例如ExecutorService和Future。 -
并不旨在为 Java 定义最终的结构化并发 API。其他的结构化并发结构可以由第三方库或在未来的 JDK 版本中定义。
-
并不旨在定义一种在线程之间共享数据流的机制（即 通道）。我们可能会在未来提出相关建议。
-
并不旨在用新的线程取消机制替换现有的线程中断机制。我们可能会在未来提出相关建议。
动机
开发者通过将任务分解为多个子任务来管理复杂性。在普通的单线程代码中，子任务是按顺序执行的。然而，如果子任务之间足够独立，并且硬件资源充足，那么通过并发执行子任务，可以加快整个任务的运行速度（即降低延迟）。例如，一个组合多个 I/O 操作结果的任务，如果每个 I/O 操作都在自己的线程中并发执行，那么该任务将运行得更快。虚拟线程（JEP 425）使得为每个这样的 I/O 操作分配一个线程变得经济高效，但管理可能产生的大量线程仍然是一个挑战。
使用 ExecutorService 的非结构化并发
Java 5 中引入的 java.util.concurrent.ExecutorService API 可帮助开发者并发执行子任务。
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 块中调用其他任务的 futures 的 cancel(boolean) 方法。我们还需要在 try-with-resources 语句 中使用 ExecutorService，正如 JEP 425 中的示例所示，因为 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() 失败，那么我们根本不会启动 fetchOrder()，并且 handle() 任务会隐式失败。子任务只能返回给其父任务这一事实非常重要：这意味着父任务可以隐式地将一个子任务的失败作为触发器，取消所有剩余的子任务，然后自身也失败。
在单线程代码中，任务与子任务的层次结构在运行时会具体化为调用栈。因此，我们可以免费获得相应的父子关系，这些关系决定了错误传播的方式。当观察单个线程时，这种层次关系是显而易见的：findUser()（以及后来的 fetchOrder()）似乎从属于 handle()。
如果任务及其子任务之间的父子关系能够像单线程代码一样在语法上表达并在运行时具体化，那么多线程编程将变得更加容易、更加可靠且更易于观察。语法结构将划定子任务的生命周期，并实现线程间层次结构的运行时表示，类似于线程内的调用栈。该表示将支持错误传播和取消操作，并允许对并发程序进行有意义的观察。
（Java 已经有一个用于对并发任务施加结构的 API，即 java.util.concurrent.ForkJoinPool，它是并行流背后的执行引擎。然而，该 API 是为计算密集型任务设计的，而不是涉及 I/O 的任务。）
结构化并发
结构化并发 是一种多线程编程方法，它保留了单线程代码的可读性、可维护性和可观测性。它体现了以下原则：
如果一个任务拆分为并发子任务，那么它们都会返回到同一个地方，即任务的代码块。
“结构化并发”一词由 Martin Sústrik 提出，并由 Nathaniel J. Smith 推广。其他语言中的思想，例如 Erlang 的分层监督器，为结构化并发中的错误处理设计提供了参考。
在结构化并发中，子任务代表一个任务工作。该任务等待子任务的结果并监控它们是否失败。与单线程代码的结构化编程技术一样，多线程的结构化并发的力量来自于两个理念：（1）通过代码块的执行流程有明确的入口和出口点，以及（2）操作生命周期的严格嵌套方式，这种方式反映了它们在代码中的语法嵌套。
由于一段代码块的入口和出口定义明确，一个并发子任务的生命周期被限制在其父任务的语法块之内。因为同级子任务的生命周期嵌套在它们父任务的生命周期之中，所以可以将它们作为一个整体来推理和管理。由于父任务的生命周期又依次嵌套在其父任务的生命周期之内，运行时可以将任务的层次结构具体化为一棵树。这棵树是单一线程调用栈的并发对应物，可观测性工具可以利用它来将子任务呈现为隶属于它们的父任务。
结构化并发非常适合虚拟线程，虚拟线程是由 JDK 实现的轻量级线程。许多虚拟线程共享同一个操作系统线程，从而允许存在大量的虚拟线程。除了数量众多外，虚拟线程还足够廉价，可以表示任何并发的行为单元，即使是涉及 I/O 的行为也可以胜任。这意味着服务器应用程序可以使用结构化并发一次处理成千上万甚至百万级别的传入请求：它可以为每个请求的处理任务分配一个新的虚拟线程，并且当任务通过提交子任务进行并发执行时，还可以为每个子任务分配一个新的虚拟线程。在幕后，任务-子任务的关系通过让每个虚拟线程携带对其唯一父线程的引用来具体化为一棵树，这类似于调用栈中的一个帧引用其唯一的调用者。
总之，虚拟线程提供了大量的线程。结构化并发确保它们被正确且稳健地协调，并使可观测性工具能够按照开发者理解的方式显示线程。在 JDK 中拥有一个结构化并发的 API 将提高服务器应用程序的可维护性、可靠性和可观测性。
描述
结构化并发 API 的主要类是 StructuredTaskScope。这个类允许开发者将一个任务组织为一组并发的子任务，并将它们作为一个单元进行协调。子任务通过单独 forking（分叉）到各自的线程中执行，然后以单元形式 joining（合并），并且可能会以单元形式取消。子任务的成功结果或异常会被父任务汇总并处理。StructuredTaskScope 将子任务（或称 forks）的生命周期限定在一个明确的词法作用域内，在此范围内完成任务与其子任务之间的所有交互 —— 包括分叉、合并、取消、错误处理以及结果组合。
以下是之前介绍的 handle() 示例，改写为使用 StructuredTaskScope（ShutdownOnFailure 将在下方解释）：
Response handle() throws ExecutionException, InterruptedException {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Future<String> user = scope.fork(() -> findUser());
Future<Integer> order = scope.fork(() -> fetchOrder());
scope.join(); // Join both forks
scope.throwIfFailed(); // ... and propagate errors
// Here, both forks have succeeded, so compose their results
return new Response(user.resultNow(), order.resultNow());
}
}
与原始示例相比，理解此处涉及的线程的生命周期非常容易：在所有条件下，它们的生命周期都被限制在一个词法作用域内，即 try-with-resources 语句的主体。此外，使用 StructuredTaskScope 确保了许多有价值的属性：
-
通过短路处理错误 — 如果
findUser()或fetchOrder()子任务中的任何一个失败，另一个任务如果尚未完成将被取消。（这是由ShutdownOnFailure实现的取消策略管理的；也可能存在其他策略）。 -
取消传播 — 如果运行
handle()的线程在调用join()之前或期间被中断，当线程退出作用域时，两个分支会自动取消。 -
清晰性 — 上述代码具有清晰的结构：设置子任务，等待它们完成或被取消，然后决定是成功（并处理已完成的子任务结果）还是失败（子任务已经结束，因此无需进一步清理）。
-
可观测性 — 如下文所述的线程转储清楚地显示了任务层次结构，其中运行
findUser()和fetchOrder()的线程显示为作用域的子线程。
与 ExecutorService.submit(...) 类似，StructuredTaskScope.fork(...) 方法接受一个 Callable 并返回一个 Future。然而，与 ExecutorService 不同的是，返回的 future 并不打算通过其 get() 方法进行合并，也不通过其 cancel() 方法取消。相反，范围内的所有分支都应作为一个整体来合并或取消。两个新的 Future 方法 resultNow() 和 exceptionNow() 旨在子任务完成后使用，例如在调用 scope.join() 之后。
使用 StructuredTaskScope
使用 StructuredTaskScope 的代码的一般工作流程如下：
-
创建一个作用域。创建该作用域的线程是其拥有者。
-
在作用域中分叉并发子任务。
-
作用域中的任何分叉或者作用域的拥有者都可以调用作用域的
shutdown()方法，以请求取消所有剩余的子任务。 -
作用域的拥有者作为一个整体加入作用域，即它的所有分叉。拥有者可以调用作用域的
join()方法，该方法会阻塞，直到所有分叉都已完成（成功或失败）或通过shutdown()被取消。或者，拥有者可以调用作用域的joinUntil(java.time.Instant)方法，该方法接受一个截止时间。 -
加入后，处理分叉中的任何错误并处理它们的结果。
-
关闭作用域，通常通过
try-with-resources 隐式关闭。这将关闭作用域并等待任何滞后的分叉完成。
如果所有者是现有作用域的成员（即，作为其中一个分支创建的），则该作用域将成为新作用域的父级。任务因此形成一棵树，作用域为中间节点，线程为叶节点。
每个分支都在其自己新创建的线程中运行，默认情况下为虚拟线程。分支的线程由作用域所拥有，而作用域又由其创建线程所拥有，从而形成一个层级结构。任何分支都可以创建自己的嵌套 StructuredTaskScope 以分支出自己的子任务，从而扩展该层级结构。这种层级结构反映在代码的块结构中，它限制了分支的生命周期：一旦作用域关闭，所有分支的线程都保证已终止，并且在块退出时不会留下任何线程。
作用域中的任何分叉、嵌套作用域中的任何分叉，以及作用域的所有者都可以在任何时候调用作用域的 shutdown() 方法来表示任务已完成 —— 即使其他分叉仍在运行。shutdown() 方法会中断作用域中所有仍在活动的分叉线程。因此，所有分叉都应该编写为能够响应中断的方式。实际上，shutdown() 是顺序代码中 break 语句的并发类比。
当 join() 返回时，所有分叉（fork）都已完成（成功或失败）或已被取消。可以通过它们的 future 的 resultNow() 或 exceptionNow() 方法获取其结果或异常，且无需额外的阻塞。（如果在 future 完成之前调用这些方法，会抛出 IllegalStateException 异常。）
在作用域内调用 join() 或 joinUntil() 是强制性的。如果作用域的代码块在 join 之前退出，那么该作用域将等待所有 fork 终止，然后抛出异常。
一个作用域的拥有线程有可能在加入之前或加入过程中被中断。例如，它可能是已关闭的封闭作用域的一个分支（fork）。如果发生这种情况，join() 和 joinUntil(Instant) 将抛出异常，因为继续执行已无意义。随后，try-with-resources 语句将关闭该作用域，这会取消所有分支并等待它们终止。这将自动将任务的取消传播到其子任务。如果 joinUntil(Instant) 方法的截止时间在分支终止或调用 shutdown() 之前到期，它将抛出异常，并且同样地，try-with-resources 语句将关闭该作用域。
StructuredTaskScope 的结构化使用在运行时强制执行。例如，尝试从不在该作用域的树层次结构中的线程调用 fork(Callable) —— 即，拥有者、分支以及嵌套作用域中的分支 —— 将因异常而失败。在 try-with-resources 块之外使用作用域，并且在未调用 close() 或未保持正确的 close() 调用嵌套顺序的情况下返回，可能会导致该作用域的方法抛出 StructureViolationException 异常。
StructuredTaskScope 对并发操作施加了结构和顺序。因此它没有实现 ExecutorService 或 Executor 接口，因为这些接口的实例通常以非结构化的方式使用（见下文）。然而，将使用 ExecutorService 的代码迁移到 StructuredTaskScope 是非常直接的，特别是那些能从结构化中受益的代码。
StructuredTaskScope 位于一个 孵化模块 中，默认情况下被排除
上面的示例使用了 StructuredTaskScope API，因此要在 JDK XX 上运行它们，必须添加 jdk.incubator.concurrent 模块，并且启用预览功能以开启虚拟线程：
-
使用
javac --release XX --enable-preview --add-modules jdk.incubator.concurrent Main.java编译程序，并使用java --enable-preview --add-modules jdk.incubator.concurrent Main运行它；或者， -
当使用 源代码启动器 时，通过
java --source XX --enable-preview --add-modules jdk.incubator.concurrent Main.java运行程序；或者， -
当使用 jshell 时，通过
jshell --enable-preview --add-modules jdk.incubator.concurrent启动它。
关机策略
在处理并发子任务时，通常会使用短路模式来避免做不必要的工作。例如，有时如果其中一个子任务失败（即，invoke all），或者其中一个子任务成功（即，invoke any），取消所有子任务是有意义的。StructuredTaskScope 的两个子类 ShutdownOnFailure 和 ShutdownOnSuccess 支持这些模式，并分别在首次分叉失败或成功时关闭作用域的策略。它们还提供了用于处理异常和成功结果的方法。
下面是一个具有故障关闭策略的 StructuredTaskScope（在上面的 handle() 示例中也使用了该策略），它会并发运行一组任务，如果其中任何一个任务失败，则整个操作失败：
<T> List<T> runAll(List<Callable<T>> tasks) throws Throwable {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
List<Future<T>> futures = tasks.stream().map(scope::fork).toList();
scope.join();
scope.throwIfFailed(e -> e); // Propagate exception as-is if any fork fails
// Here, all tasks have succeeded, so compose their results
return futures.stream().map(Future::resultNow).toList();
}
}
这是一个带有成功后关闭策略的 StructuredTaskScope，它会返回第一个成功子任务的结果：
<T> T race(List<Callable<T>> tasks, Instant deadline) throws ExecutionException {
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<T>()) {
for (var task : tasks) {
scope.fork(task);
}
scope.joinUntil(deadline);
return scope.result(); // Throws if none of the forks completed successfully
}
}
一旦某个分支成功，此作用域会自动关闭，并取消剩余的活动分支。如果所有分支都失败或者超过了给定的截止时间，任务就会失败。例如，在需要从一组冗余服务中获取结果的服务器应用程序中，这种模式可能会很有用。
虽然提供了这两种关闭策略，但开发者可以通过扩展 StructuredTaskScope 并重写 handleComplete(Future) 方法来创建抽象其他模式的自定义策略。
扇入场景
上面的示例聚焦于扇出场景，该场景管理多个并发的传出 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 425 添加的新的 JSON 线程转储格式，以显示 StructuredTaskScope 将线程分组为层次结构的方式：
$ jcmd <pid> Thread.dump_to_file -format=json <file>
每个作用域的 JSON 对象包含该作用域中分叉出的线程及其堆栈跟踪的数组。一个作用域的拥有线程通常会阻塞在 join 方法中，等待子任务完成；线程转储通过显示结构化并发所施加的树层次结构，可以轻松查看子任务线程正在执行的操作。作用域的 JSON 对象还包含对其父作用域的引用，从而可以从转储中重建程序的结构。
com.sun.management.HotSpotDiagnosticsMXBean API 也可以用来生成此类线程转储，可以直接生成，也可以通过平台 MBeanServer 以及本地或远程 JMX 工具间接生成。
替代方案
-
什么都不做。让开发者继续使用现有的低级
java.util.concurrentAPI，并继续仔细考虑并发代码中出现的所有异常情况和生命周期协调问题。 -
增强
ExecutorService接口。我们为该接口制作了一个原型实现，该实现始终强制执行结构并限制哪些线程可以提交任务。然而，我们发现这存在问题，因为在 JDK 和生态系统中，大多数ExecutorService（及其父接口Executor）的使用都不是结构化的。为一个限制性更强的概念重用相同的 API 必然会引起混淆。例如，在大多数情况下，将一个结构化的ExecutorService实例传递给接受此类型的现有方法几乎肯定会抛出异常。