Java新的结构化并行模式入门指南
译者 | 布加迪
结构化并发是JAVA中使用多线程的一种新方式。它允许开发人员在充分利用传统线程和虚拟线程的同时考虑逻辑组中的工作。结构化并发出现在Java 21的预览版中，它是决定Java未来的一个关键方面，所以现在是开始使用它的好时机。
为什么我们需要结构化并发？
编写并发软件是软件开发者面临的最大挑战之一。Java的线程模式使其成为并发语言中的有力竞争者，但是多线程一直天生很棘手。结构化并发允许您使用具有结构化编程语法的多线程。实质上，它提供了一种使用熟悉的程序流程和构件编写并发软件的方法。这让开发者可以专注于手头的事务，而不是线程编排。正如结构化并发性的JEP所说：“如果一个任务分成并发子任务，它们都回到相同的位置，即任务的代码块。”
虚拟线程现在是Java的一项正式特性，它可以低成本生成线程，从而获得并发性能。结构化并发提供了这么做的简单语法。因此，Java现在有了一个独特的、高度优化的线程系统，而且易于理解。
新的StructuredTaskScope类
结构化并发中的主要类是java.util.concurrent.StructuredTaskScope。Java 21文档包含如何使用结构化并发的示例。截止本文发稿时，您需要使用--enable-preview和--source 21或--source 22来启用Java程序中的结构化并发。我的$java --version是openjdk 22-ea，所以我们使用Maven的示例将为编译步骤指定--enable-preview --source 22，为执行步骤指定--enable-preview。注意，SDKMan对于管理多个JDK安装是一个很好的选择。
您可以在本文的Github代码存储库中找到示例代码。注意为执行设置—enable-preview的.mvn/jvm.config文件。若要运行代码，使用$mvn clean compile exec:java。
具有结构化并发的多线程
就本文示例而言，我们将向Star Wars API（SWAPI）发出几个请求，通过行星的ID获取有关行星的信息。如果我们在标准的同步Java中执行此操作，可能会使用Apache HTTPClient执行类似代码片段1的操作。
代码片段1. 类似传统方法的多API调用
复制
package com.infoworld;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
public class App {
public String getPl.NET(int planetId) throws Exception {
System.out.println("BEGIN getPlanet()");
String url = "https://swapi.dev/api/planets/" + planetId + "/";
String ret = "?";
CloseableHttpClient httpClient = HttpClients.createDefault();
HttpGet request = new HttpGet(url);
CloseableHttpResponse response = httpClient.execute(request);
// Check the response status code
if (response.getStatusLine().getStatusCode() != 200) {
System.err.println("Error fetching planet information for ID: " + planetId);
throw new RuntimeException("Error fetching planet information for ID: " + planetId);
} else {
// Parse the JSON response and extract planet information
ret = EntityUtils.toString(response.getEntity());
System.out.println("Got a Planet: " + ret);
}
// Close the HTTP response and client
response.close();
httpClient.close();
return ret;
}
void sync() throws Exception {
int[] planetIds = {1,2,3,4,5};
for (int planetId : planetIds) {
getPlanet(planetId);
}
}
public static void mAIn(String[] args) {
var myApp = new App();
System.out.println("nr-- BEGIN Sync");
try {
myApp.sync();
} catch (Exception e){
System.err.println("Error: " + e);
}
}
}
在代码片段1中，我们有一个调用sync()方法的主方法，该方法在调用“https://swapi.dev/api/planets/”+ planetId端点时，只是对一组ID进行迭代处理。这些调用通过getPlanet()方法发出，该方法使用Apache HTTP库来处理样板请求、响应和错误处理。实际上，该方法接收每个响应，如果正确（200），输出到控制台，否则抛出错误。（这些示例使用了最少的错误，所以在这种情况下我们只抛出RuntimeException。）
输出是这样的：
复制
-- BEGIN Sync
BEGIN getPlanet()
Got a Planet: {"name":"Tatooine"}
BEGIN getPlanet()
Got a Planet: {"name":"Alderaan"}
BEGIN getPlanet()
Got a Planet: {"name":"Yavin”}
BEGIN getPlanet()
Got a Planet: {"name":"Hoth"}
BEGIN getPlanet()
Got a Planet: {"name":"Dagobah"}
现在不妨使用结构化并发尝试同一个示例。如代码片段2所示，结构化并发允许我们将调用分解成并发请求，并将所有内容放在相同的代码空间中。在代码片段2中，我们添加了必要的StructuredTaskScope导入，然后使用其核心方法fork()和join()，将每个请求分解成各自的线程，然后等待它们全部完成。
代码片段2. 使用StructuredTaskScopeNow的多API调用
复制
package com.infoworld;
import java.util.concurrent.*;
import java.util.concurrent.StructuredTaskScope.*;
//...
public class App {
public String getPlanet(int planetId) throws Exception {
// ... same ...
}
void sync() throws Exception {
int[] planetIds = {1,2,3,4,5};
for (int planetId : planetIds) {
getPlanet(planetId);
}
}
void sc() throws Exception {
int[] planetIds = {1,2,3,4,5};
try (var scope = new StructuredTaskScope<Object>()) {
for (int planetId : planetIds) {
scope.fork(() -> getPlanet(planetId));
}
scope.join();
}catch (Exception e){
System.out.println("Error: " + e);
}
}
public static void main(String[] args) {
var myApp = new App();
// ...
System.out.println("nr-- BEGIN Structured Concurrency");
try {
myApp.sc();
} catch (Exception e){
System.err.println("Error: " + e);
}
}
}
如果我们运行代码片段2，将得到类似的输出，但速度要快不少，这是由于请求是同时发出、并发进行的。不妨考虑sc()方法（使用多线程）与sync()方法（使用同步代码）之间的区别。结构化并发方法没有想象的那么难，提供结果的速度却快得多。
处理任务和子任务
默认情况下，StructuredTaskScope被创建时，它使用虚拟线程，所以我们实际上并没有在这里配置操作系统线程；相反，我们告诉JVM以最有效的方式编排请求。（StructuredTaskScope的构造函数也接受ThreadFactory。）
在代码片段2中，我们在try-with-resource块中创建StructuredTaskScope对象，这是它原本的使用方式。我们可以使用fork()创建任意数量的作业。fork()方法接受任何实现Callable的程序，也就是说，任何方法或函数。这里，我们将getPlanet()方法包装在一个匿名函数中：()-> getPlanet(planetId)——这是一种向目标函数传递参数的实用语法。
当我们调用join()时，我们告诉作用域等待所有被分叉的作业。实质上，join()将我们带回到同步模式。分叉的作业将按照TaskScope的配置进行处理。
关闭任务作用域
由于我们在try-with-resource块中创建了TaskScope，因此当该块结束时，作用域将自动关闭。这为作用域调用shutdown()进程，作用域可以定制，以便根据需要来处理运行中线程的处置。如果需要在作用域关闭之前关闭它，也可以手动调用shutdown()方法。
StructuredTaskScope包括两个实现内置关闭策略的类：ShutDownOnSuccess和ShutDownOnFailure。这些类监视成功或出错的子任务，然后取消其余运行中的线程。使用目前的设置，我们可以这样使用这些类：
代码片段3. 内置关闭策略
复制
void failFast() throws ExecutionException, InterruptedException {
int[] planetIds = {1,2,3,-1,4};
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
for (int planetId : planetIds) {
scope.fork(() -> getPlanet(planetId));
}
scope.join();
}
}
void succeedFast() throws ExecutionException, InterruptedException {
int[] planetIds = {1,2};
try (var scope = new StructuredTaskScope.ShutdownOnSuccess()) {
for (int planetId : planetIds) {
scope.fork(() -> getPlanet(planetId));
}
scope.join();
} catch (Exception e){
System.out.println("Error: " + e);
}
}
public static void main(String[] args) {
var myApp = new App();
System.out.println("nr-- BEGIN succeedFast");
try {
myApp. succeedFast();
} catch (Exception e) {
System.out.println(e.getMessage());
}
System.out.println("nr-- BEGIN failFast");
try {
myApp.failFast();
} catch (Exception e) {
System.out.println(e.getMessage());
}
}
这些策略将给出类似以下的输出：
复制
-- BEGIN succeedFast
BEGIN getPlanet()
BEGIN getPlanet()
Got a Planet: {"name":"Alderaan"}
org.apache.http.impl.execchain.RetryExec execute
INFO: I/O exception (java.net.SocketException) caught when processing request to {s}->https://swapi.dev:443: Closed by interrupt
-- BEGIN failFast
BEGIN getPlanet()
BEGIN getPlanet()
BEGIN getPlanet()
BEGIN getPlanet()
BEGIN getPlanet()
Got a Planet: {"name":"Hoth"}
Got a Planet: {"name":"Tatooine"}
Error fetching planet information for ID: -1
org.apache.http.impl.execchain.RetryExec execute
INFO: I/O exception (java.net.SocketException) caught when processing request to {s}->https://swapi.dev:443: Closed by interrupt
因此，我们拥有的是一种简单的机制，可以并发启动所有请求，然后在一个请求成功或失败时取消其余的请求。这里，可以进行任何定制。结构化并发文档包括一个示例，在子任务成功或失败时收集子任务结果，然后返回结果。这很容易完成，只需通过覆盖join()方法，并观察每个任务的结果。
StructuredTaskScope.Subtask
在我们的示例中没有看到的一件事是观察子任务的返回值。每次StructuredTaskScope.fork()被调用时，就返回StructuredTaskScope.SubTask对象。我们可以利用它来观察任务的状态。比如在sc()方法中，我们可以这么做：
代码片段4. 使用StructuredTaskScope.Subtask观察状态
复制
import java.util.concurrent.StructuredTaskScope.Subtask;
import java.util.ArrayList;
void sc() throws Exception {
int[] planetIds = {1,2,3,4,5};
ArrayList<Subtask> tasks = new ArrayList<Subtask>(planetIds.length);
try (var scope = new StructuredTaskScope<Object>()) {
for (int planetId : planetIds) {
tasks.add(scope.fork(() -> getPlanet(planetId)));
}
scope.join();
}catch (Exception e){
System.out.println("Error: " + e);
}
for (Subtask t : tasks){
System.out.println("Task: " + t.state());
}
}
在这个示例中，我们将每个任务保存在ArrayList中，然后在进行join()操作之后输出它们的状态。注意，Subtask的可用状态被定义为enum。这个新方法将输出类似以下的内容：
复制
-- BEGIN Structured Concurrency
BEGIN getPlanet()
BEGIN getPlanet()
BEGIN getPlanet()
BEGIN getPlanet()
BEGIN getPlanet()
Got a Planet: {"name":"Dagobah"}
Got a Planet: {"name":"Hoth"}
Got a Planet: {"name":"Tatooine"}
Got a Planet: {"name":"Yavin IV"}
Got a Planet: {"name":"Alderaan"}
Task: SUCCESS
Task: SUCCESS
Task: SUCCESS
Task: SUCCESS
Task: SUCCESS
结论
在虚拟线程和结构化并发之间，Java开发者拥有一种引人注目的新机制，可以将几乎所有代码分解成并发任务，不会有太大的开销。上下文和需求很重要，所以不要仅仅因为存在这些新的并发工具就使用它们。与此同时，这种组合确实提供了一些强大的力量。一旦您遇到出现许多任务的瓶颈时，您可以轻松地将它们全部交给虚拟线程引擎，该引擎将找到编排它们的最佳方法。具有结构化并发的新线程模式还使您易于定制和微调这种行为。
至于开发者将来如何在我们的应用程序、框架和服务器中使用这些新的并发功能，值得我们拭目以待。
小知识：结构化并发中的线程树
结构化并发包括对调试和理解线程之间关系的支持。特别是，结构化并发将所有线程关联到树结构中，作用域位于根。这样一来，查看线程之间的关系就变得很简单，即便使用嵌套作用域也是如此。说明文档提供了一个好的示例，表明如何使用Java诊断命令（jcmd）实用程序，将线程的运行时布局转储到控制台。
原文标题：Get started with Java's new structured concurrency model，作者：Matthew Tyson
