Java结构化并发编程怎么用

wen java案例 2026-07-10 192

Java结构化并发编程使用指南

结构化并发是Java 21引入的重要特性，通过StructuredTaskScope实现，下面详细介绍其使用方法。

Java结构化并发编程怎么用

基本概念

结构化并发将并发任务的生命周期与代码块绑定,确保所有子任务在代码块结束前完成。

基础使用

最简单的例子

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";
    }
}