文章详情

短信预约-IT技能 免费直播动态提醒

请输入下面的图形验证码

提交验证

短信预约提醒成功

Java新的结构化并行模式入门指南

2024-11-30 03:47

关注

审校 | 重楼

结构化并发是Java中使用多线程的一种新方式。它允许开发人员在充分利用传统线程和虚拟线程的同时考虑逻辑组中的工作。结构化并发出现在Java 21的预览版中,决定Java未来的一个关键方面,所以现在是开始使用它的好时机。

为什么我们需要结构化并发

编写并发软件是软件开发面临的最大挑战之一。Java的线程模使其成为并发语言中的有力竞争者,但是多线程一直天生很棘手。结构化并发允许您使用具有结构化编程语法的多线程。实质上,它提供了一种使用熟悉的程序流程和构编写并发软件的方法。这让开发可以专注于手头的事务,而不是线程编排。正如结构化并发性的JEP所说:“如果一个任务分成并发子任务,它们都回到相同的位置,即任务的代码块。

虚拟线程现在是Java的一正式特性,它可以低成本生成线程,从而获得并发性能。结构化并发提供了这做的简单语法。因此,Java现在有了一个独特的、高度优化的线程系统,而且易于理解

新的StructuredTaskScope类

结构化并发中的主要类是java.util.concurrent.StructuredTaskScopeJava 21文档包含如何使用结构化并发的示例。截止本文发稿时,您需要使用--enable-preview--source 21--source 22来启用Java程序中的结构化并发。我的$java --versionopenjdk 22-ea,所以我们使用Maven例将为编译步骤指定--enable-preview --source 22,为执行步骤指定--enable-preview。注意,SDKMan对于管理多个JDK安装是一个很好的选择。

您可以在本文的GitHub代码存储库中找到示例代码。注意为执行设置enable-preview的.mvn/jvm.config文件。若要运行代码,使用$mvn clean compile exec:java

具有结构化并发的多线程

就本文示例而言,我们将向Star Wars APISWAPI发出几个请求,通过行星的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 getPlanet(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("\n\r-- 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()) {
 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("\n\r-- 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("\n\r-- BEGIN succeedFast");
 try {
 myApp. succeedFast();
 } catch (Exception e) {
 System.out.println(e.getMessage());
 } 
 System.out.println("\n\r-- 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 tasks = new ArrayList(planetIds.length); 
 try (var scope = new StructuredTaskScope()) {
 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


免责声明:

① 本站未注明“稿件来源”的信息均来自网络整理。其文字、图片和音视频稿件的所属权归原作者所有。本站收集整理出于非商业性的教育和科研之目的,并不意味着本站赞同其观点或证实其内容的真实性。仅作为临时的测试数据,供内部测试之用。本站并未授权任何人以任何方式主动获取本站任何信息。

② 本站未注明“稿件来源”的临时测试数据将在测试完成后最终做删除处理。有问题或投稿请发送至: 邮箱/279061341@qq.com QQ/279061341

软考中级精品资料免费领

  • 历年真题答案解析
  • 备考技巧名师总结
  • 高频考点精准押题
  • 资料下载
  • 历年真题
  • 2024年上半年信息系统项目管理师第二批次真题及答案解析(完整版)

    难度     813人已做
    查看
  • 【考后总结】2024年5月26日信息系统项目管理师第2批次考情分析

    难度     354人已做
    查看
  • 【考后总结】2024年5月25日信息系统项目管理师第1批次考情分析

    难度     318人已做
    查看
  • 2024年上半年软考高项第一、二批次真题考点汇总(完整版)

    难度     435人已做
    查看
  • 2024年上半年系统架构设计师考试综合知识真题

    难度     224人已做
    查看

相关文章

发现更多好内容
咦!没有更多了?去看看其它编程学习网 内容吧