Java结构化并发编程使用指南
结构化并发是Java 21引入的重要特性，通过StructuredTaskScope实现，下面详细介绍其使用方法。
基本概念
结构化并发将并发任务的生命周期与代码块绑定,确保所有子任务在代码块结束前完成。
基础使用
最简单的例子
import java.util.concurrent.StructuredTaskScope;
public class BasicStructuredConcurrency {
public static void main(String[] args) throws Exception {
// 创建结构化任务作用域
try (var scope = new StructuredTaskScope<String>()) {
// 提交子任务
StructuredTaskScope.Subtask<String> task1 = scope.fork(() -> {
Thread.sleep(1000);
return "Task 1 result";
});
StructuredTaskScope.Subtask<String> task2 = scope.fork(() -> {
Thread.sleep(2000);
return "Task 2 result";
});
// 等待所有任务完成
scope.join();
// 获取结果
System.out.println(task1.get());
System.out.println(task2.get());
}
}
}
常用策略模式
1 全部成功策略 (ShutdownOnSuccess)
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.ExecutionException;
public class ShutdownOnSuccessExample {
public static String fetchFromMultipleServices() throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
// 并行查询多个服务
scope.fork(() -> queryService("ServiceA"));
scope.fork(() -> queryService("ServiceB"));
scope.fork(() -> queryService("ServiceC"));
// 返回第一个成功的结果，其他任务会被取消
return scope.join().result();
}
}
private static String queryService(String service) {
// 模拟服务调用
return service + ": result";
}
}
2 全部失败策略 (ShutdownOnFailure)
import java.util.concurrent.StructuredTaskScope;
public class ShutdownOnFailureExample {
public static void processAllTasks() throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
StructuredTaskScope.Subtask<String> task1 = scope.fork(() -> {
return doWork("Task1");
});
StructuredTaskScope.Subtask<Integer> task2 = scope.fork(() -> {
return calculate("Task2");
});
// 等待，如果任意任务失败则抛出异常
scope.join().throwIfFailed();
// 所有任务成功，获取结果
System.out.println(task1.get());
System.out.println(task2.get());
}
}
private static String doWork(String name) {
return "Completed: " + name;
}
private static int calculate(String name) {
return 42;
}
}
实际应用示例
1 并发数据聚合
import java.util.concurrent.StructuredTaskScope;
import java.util.List;
import java.util.ArrayList;
public class DataAggregationExample {
// 聚合多个数据源的结果
public static AggregatedResult aggregateData(int userId) throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
// 并行获取各种数据
StructuredTaskScope.Subtask<UserProfile> profileTask = scope.fork(() ->
fetchUserProfile(userId));
StructuredTaskScope.Subtask<List<Order>> ordersTask = scope.fork(() ->
fetchUserOrders(userId));
StructuredTaskScope.Subtask<Preferences> prefsTask = scope.fork(() ->
fetchUserPreferences(userId));
// 等待所有任务完成
scope.join().throwIfFailed();
// 聚合结果
return new AggregatedResult(
profileTask.get(),
ordersTask.get(),
prefsTask.get()
);
}
}
// 模拟数据获取方法
private static UserProfile fetchUserProfile(int id) {
return new UserProfile(id, "User" + id);
}
private static List<Order> fetchUserOrders(int id) {
return List.of(new Order("Order1"), new Order("Order2"));
}
private static Preferences fetchUserPreferences(int id) {
return new Preferences("dark_mode", "en");
}
// 数据类
record UserProfile(int id, String name) {}
record Order(String name) {}
record Preferences(String theme, String language) {}
record AggregatedResult(UserProfile profile, List<Order> orders, Preferences preferences) {}
}
2 超时控制
import java.util.concurrent.StructuredTaskScope;
import java.time.Duration;
import java.time.Instant;
public class TimeoutExample {
public static String fetchWithTimeout() throws Exception {
Instant deadline = Instant.now().plusSeconds(3);
try (var scope = new StructuredTaskScope<String>()) {
scope.fork(() -> {
Thread.sleep(5000); // 模拟慢请求
return "Slow result";
});
scope.fork(() -> {
Thread.sleep(1000); // 模拟快请求
return "Fast result";
});
// 等待到期或所有任务完成
while (Instant.now().isBefore(deadline)) {
if (scope.isDone()) {
break;
}
Thread.sleep(100);
}
scope.join();
// 处理已完成的任务
scope.stream()
.filter(StructuredTaskScope.Subtask::isCompletedSuccessfully)
.findFirst()
.ifPresent(task -> System.out.println("Result: " + task.get()));
return "Completed";
}
}
}
错误处理最佳实践
import java.util.concurrent.StructuredTaskScope;
public class ErrorHandlingExample {
public static void robustConcurrentExecution() throws Exception {
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
scope.fork(() -> {
try {
return riskyOperation1();
} catch (Exception e) {
System.err.println("Operation 1 failed: " + e.getMessage());
throw e;
}
});
scope.fork(() -> {
try {
return riskyOperation2();
} catch (Exception e) {
System.err.println("Operation 2 failed: " + e.getMessage());
throw e;
}
});
try {
scope.join().throwIfFailed();
} catch (Exception e) {
System.err.println("At least one task failed: " + e.getMessage());
// 可以执行清理或降级逻辑
throw e;
}
}
}
private static String riskyOperation1() {
if (Math.random() > 0.5) {
throw new RuntimeException("Random failure");
}
return "Success 1";
}
private static String riskyOperation2() {
if (Math.random() > 0.5) {
throw new RuntimeException("Random failure");
}
return "Success 2";
}
}
关键注意事项
- 线程继承：子任务会继承父线程的上下文
- 自动取消：任务作用域关闭时，未完成的任务会被自动取消
- 异常传播：异常会正确传播到父作用域
- 资源管理：使用try-with-resources确保作用域正确关闭
结构化并发提供了一种更安全、更易理解的并发编程模型：
- ✓ 自动管理线程生命周期
- ✓ 清晰的错误处理
- ✓ 避免线程泄漏
- ✓ 代码更易维护
- ✓ 与虚拟线程完美配合
使用结构化并发可以让你的并发代码更可靠、更容易调试。
