Java 25 虚拟线程与结构化并发的结合：并发编程的黄金组合-程序员充电站

Java 25 虚拟线程与结构化并发的结合：并发编程的黄金组合

别叫我大神，叫我 Alex 就好。今天我们来聊聊 Java 25 中虚拟线程与结构化并发的结合，这是并发编程的黄金组合。

一、虚拟线程与结构化并发的关系

虚拟线程和结构化并发是 Java 近年来引入的两个革命性特性，它们解决了不同但相关的问题：

虚拟线程：解决了传统线程的资源占用高、创建成本高的问题，让我们可以轻松创建数百万个线程
结构化并发：解决了并发任务的管理问题，让我们可以更清晰地管理任务的生命周期

当这两个特性结合使用时，它们形成了一个强大的组合，让并发编程变得更加简单、优雅和高效。

二、基本用法

1. 在结构化并发中使用虚拟线程

public class VirtualThreadsWithStructuredConcurrency { public User getUserWithDetails(String userId) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { // 使用虚拟线程执行任务 Future<User> userFuture = scope.fork(() -> { return userService.getUser(userId); }); Future<UserProfile> profileFuture = scope.fork(() -> { return profileService.getUserProfile(userId); }); Future<List<Order>> ordersFuture = scope.fork(() -> { return orderService.getUserOrders(userId); }); scope.join(); scope.throwIfFailed(); User user = userFuture.resultNow(); user.setProfile(profileFuture.resultNow()); user.setOrders(ordersFuture.resultNow()); return user; } } }

2. 自定义虚拟线程执行器

public class VirtualThreadExecutor { private static final Executor VIRTUAL_THREAD_EXECUTOR = Executors.newVirtualThreadPerTaskExecutor(); public static <T> T executeWithVirtualThread(Supplier<T> task) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { Future<T> future = scope.fork(task::get); scope.join(); scope.throwIfFailed(); return future.resultNow(); } } public static void executeWithVirtualThread(Runnable task) { VIRTUAL_THREAD_EXECUTOR.execute(task); } }

三、高级用法

1. 嵌套结构化并发

public class NestedStructuredConcurrency { public OrderDetails getOrderDetails(String orderId) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { // 第一层：获取订单基本信息 Future<Order> orderFuture = scope.fork(() -> { return orderService.getOrder(orderId); }); // 第二层：获取订单相关信息 Future<OrderRelatedInfo> relatedInfoFuture = scope.fork(() -> { try (var innerScope = new StructuredTaskScope.ShutdownOnFailure()) { Future<List<OrderItem>> itemsFuture = innerScope.fork(() -> { return orderItemService.getOrderItems(orderId); }); Future<Payment> paymentFuture = innerScope.fork(() -> { return paymentService.getOrderPayment(orderId); }); Future<Logistics> logisticsFuture = innerScope.fork(() -> { return logisticsService.getOrderLogistics(orderId); }); innerScope.join(); innerScope.throwIfFailed(); return new OrderRelatedInfo( itemsFuture.resultNow(), paymentFuture.resultNow(), logisticsFuture.resultNow() ); } }); scope.join(); scope.throwIfFailed(); return new OrderDetails( orderFuture.resultNow(), relatedInfoFuture.resultNow() ); } } }

2. 超时处理

public class TimeoutHandling { public User getUserWithTimeout(String userId, Duration timeout) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { Future<User> userFuture = scope.fork(() -> { return userService.getUser(userId); }); // 等待结果或超时 scope.join(timeout); if (scope.isCompletedSuccessfully()) { return userFuture.resultNow(); } else { throw new TimeoutException("Operation timed out"); } } } }

四、性能优化

1. 批量操作

public class BatchProcessing { public List<User> getUsersBatch(List<String> userIds) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { List<Future<User>> futures = userIds.stream() .map(userId -> scope.fork(() -> userService.getUser(userId))) .collect(Collectors.toList()); scope.join(); scope.throwIfFailed(); return futures.stream() .map(Future::resultNow) .collect(Collectors.toList()); } } }

2. 背压控制

public class BackpressureControl { public List<User> getUsersWithBackpressure(List<String> userIds, int batchSize) { List<User> results = new ArrayList<>(); // 分批处理，控制并发度 for (int i = 0; i < userIds.size(); i += batchSize) { int end = Math.min(i + batchSize, userIds.size()); List<String> batchIds = userIds.subList(i, end); try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { List<Future<User>> futures = batchIds.stream() .map(userId -> scope.fork(() -> userService.getUser(userId))) .collect(Collectors.toList()); scope.join(); scope.throwIfFailed(); futures.stream() .map(Future::resultNow) .forEach(results::add); } } return results; } }

五、实践案例：电商平台订单处理

场景描述

电商平台需要处理订单创建、库存检查、支付处理等多个并发操作。

实现方案

@Service public class OrderProcessingService { private final InventoryService inventoryService; private final PaymentService paymentService; private final OrderRepository orderRepository; private final EventPublisher eventPublisher; // 构造函数省略 public Order createOrder(CreateOrderRequest request) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { // 1. 检查库存 Future<Boolean> stockCheckFuture = scope.fork(() -> { return inventoryService.checkStock( request.getProductId(), request.getQuantity()); }); // 2. 验证用户 Future<User> userFuture = scope.fork(() -> { return userService.getUser(request.getUserId()); }); // 3. 计算价格 Future<BigDecimal> priceFuture = scope.fork(() -> { return pricingService.calculatePrice( request.getProductId(), request.getQuantity()); }); scope.join(); scope.throwIfFailed(); // 检查库存 if (!stockCheckFuture.resultNow()) { throw new InsufficientStockException("Not enough stock"); } // 创建订单 Order order = Order.builder() .userId(request.getUserId()) .productId(request.getProductId()) .quantity(request.getQuantity()) .totalPrice(priceFuture.resultNow()) .status(OrderStatus.CREATED) .build(); // 保存订单 order = orderRepository.save(order); // 4. 处理支付 scope.fork(() -> { PaymentRequest paymentRequest = PaymentRequest.builder() .orderId(order.getId()) .userId(request.getUserId()) .amount(priceFuture.resultNow()) .build(); return paymentService.processPayment(paymentRequest); }); // 5. 扣减库存 scope.fork(() -> { return inventoryService.decreaseStock( request.getProductId(), request.getQuantity()); }); scope.join(); scope.throwIfFailed(); // 发布订单创建事件 eventPublisher.publishEvent(new OrderCreatedEvent(order)); return order; } } }

六、常见问题与解决方案

1. 异常处理

public class ExceptionHandling { public User getUserWithErrorHandling(String userId) { try (var scope = new StructuredTaskScope.ShutdownOnFailure()) { Future<User> userFuture = scope.fork(() -> { try { return userService.getUser(userId); } catch (UserNotFoundException e) { // 处理特定异常 logger.warn("User not found: {}", userId); return new User(userId, "Unknown", "unknown@example.com"); } }); scope.join(); try { scope.throwIfFailed(); } catch (Exception e) { // 处理其他异常 logger.error("Failed to get user: {}", e.getMessage()); return new User(userId, "Error", "error@example.com"); } return userFuture.resultNow(); } } }

2. 资源管理

public class ResourceManagement { public List<User> getUsersWithResourceManagement(List<String> userIds) { try (var scope = new StructuredTaskScope.ShutdownOnFailure(); var connection = dataSource.getConnection()) { List<Future<User>> futures = userIds.stream() .map(userId -> scope.fork(() -> { try (var statement = connection.prepareStatement( "SELECT * FROM users WHERE id = ?")) { statement.setString(1, userId); try (var resultSet = statement.executeQuery()) { if (resultSet.next()) { return new User( resultSet.getString("id"), resultSet.getString("name"), resultSet.getString("email") ); } else { throw new UserNotFoundException(userId); } } } })) .collect(Collectors.toList()); scope.join(); scope.throwIfFailed(); return futures.stream() .map(Future::resultNow) .collect(Collectors.toList()); } } }