Java 21结构化并发在Project Reactor Mono.subscribe()中非结构化异常？StructuredTaskScope.adapter与Hooks.onOperatorError - 智猿学院-前后端，数据库，人工智能，云计算等领域前沿技术讲座

Java 21 结构化并发与 Project Reactor Mono.subscribe() 中的异常处理

大家好，今天我们来深入探讨 Java 21 的结构化并发特性，以及它与 Project Reactor 的 Mono.subscribe() 结合使用时可能遇到的非结构化异常问题。我们会着重讨论 StructuredTaskScope.adapter 的作用，并分析其与 Reactor 的 Hooks.onOperatorError 之间的关系。

结构化并发简介

结构化并发是 Java 21 引入的一项重要特性，它旨在改善并发编程的可靠性和可维护性。其核心思想是将并发任务的生命周期限制在一个结构化的代码块内，类似于结构化编程中的 try-catch 块。这意味着父线程可以更好地控制和管理其子线程，避免出现“线程泄露”等问题。

结构化并发主要依赖于以下几个核心类：

StructuredTaskScope: 用于管理一组并发任务的生命周期。它可以确保所有子任务完成后才能继续执行父任务，并提供取消所有任务的功能。
Thread.startVirtualThread(Runnable): 用于启动虚拟线程，虚拟线程是一种轻量级的线程，可以大大提高并发程序的性能。
Future: 用于获取异步任务的结果，并可以取消任务。

一个典型的结构化并发示例代码如下：

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.StructuredTaskScope;

public class StructuredConcurrencyExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Future<String> userFuture = scope.fork(() -> findUser());
            Future<Integer> orderFuture = scope.fork(() -> fetchOrderCount());

            scope.join(); // 等待所有任务完成
            scope.throwIfFailed(); // 如果任何任务失败，则抛出异常

            String user = userFuture.resultNow();
            Integer orderCount = orderFuture.resultNow();

            System.out.println("User: " + user + ", Order Count: " + orderCount);
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }

    static String findUser() {
        // 模拟耗时操作
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
        return "John Doe";
    }

    static Integer fetchOrderCount() {
        // 模拟耗时操作
        try {
            Thread.sleep(50);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
        return 10;
    }
}

在这个例子中，StructuredTaskScope 确保了 findUser 和 fetchOrderCount 这两个任务都会执行完毕，或者在其中一个任务失败时，另一个任务会被取消。throwIfFailed() 方法会检查是否有任务失败，如果失败，则抛出异常，从而保证了程序的健壮性。

Project Reactor 与 Mono.subscribe()

Project Reactor 是一个响应式编程框架，它基于反应式流规范，提供了强大的异步和非阻塞编程能力。Mono 是 Reactor 中的一个核心类，它代表一个包含零个或一个元素的异步序列。Mono.subscribe() 方法用于订阅这个序列，并指定处理序列元素、错误和完成信号的回调函数。

Mono.subscribe() 方法有多个重载版本，最常用的版本包括：

subscribe(Consumer<? super T> consumer): 只处理序列中的元素。
subscribe(Consumer<? super T> consumer, Consumer<? super Throwable> errorConsumer): 处理序列中的元素和错误。
subscribe(Consumer<? super T> consumer, Consumer<? super Throwable> errorConsumer, Runnable completeConsumer): 处理序列中的元素、错误和完成信号。
subscribe(Consumer<? super T> consumer, Consumer<? super Throwable> errorConsumer, Runnable completeConsumer, Consumer<? super Subscription> subscriptionConsumer): 处理序列中的元素、错误、完成信号和订阅本身。

一个简单的 Mono.subscribe() 示例代码如下：

import reactor.core.publisher.Mono;

public class MonoSubscribeExample {

    public static void main(String[] args) {
        Mono.just("Hello Reactor")
            .map(String::toUpperCase)
            .subscribe(
                System.out::println, // onNext
                Throwable::printStackTrace, // onError
                () -> System.out.println("Completed") // onComplete
            );
    }
}

在这个例子中，Mono.just() 创建了一个包含字符串 "Hello Reactor" 的 Mono 序列。map() 操作符将字符串转换为大写。subscribe() 方法定义了三个回调函数，分别用于处理序列中的元素（打印到控制台）、错误（打印堆栈信息）和完成信号（打印 "Completed"）。

结构化并发与 Mono.subscribe() 的结合

当我们将结构化并发与 Mono.subscribe() 结合使用时，可能会遇到一些问题，特别是涉及到异常处理方面。一个常见的场景是，在 Mono 序列的处理过程中发生异常，而这个异常没有被 subscribe() 方法的 errorConsumer 捕获，导致异常逃逸到结构化并发的范围之外，从而破坏了结构化并发的结构性。

考虑以下代码：

import reactor.core.publisher.Mono;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.StructuredTaskScope;

public class StructuredConcurrencyReactorExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            Future<String> reactorFuture = scope.fork(() -> {
                Mono<String> mono = Mono.just("Initial Value")
                    .map(s -> {
                        if (s.equals("Initial Value")) {
                            throw new RuntimeException("Simulated Error in Reactor");
                        }
                        return s.toUpperCase();
                    });

                // 没有 errorConsumer，异常会逃逸
                mono.subscribe(System.out::println);
                return "Reactor Task Completed";
            });

            scope.join();
            scope.throwIfFailed();

            System.out.println(reactorFuture.resultNow());
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }
}

在这个例子中，Mono 序列中的 map() 操作符会抛出一个 RuntimeException。由于 subscribe() 方法没有提供 errorConsumer，这个异常不会被 Reactor 捕获，而是会逃逸到 StructuredTaskScope 的范围之外。这会导致 StructuredTaskScope 无法正确地检测到任务失败，从而破坏了结构化并发的结构性。

StructuredTaskScope.adapter 的作用

为了解决上述问题，Java 21 提供了 StructuredTaskScope.adapter 方法。这个方法可以将一个 ExecutorService 适配成一个 StructuredTaskScope，从而允许我们在结构化并发的范围内使用现有的并发工具，例如 Reactor。

StructuredTaskScope.adapter 的主要作用是将 ExecutorService 提交的任务的异常，转换为 StructuredTaskScope 可以识别的异常，从而保证结构化并发的结构性。

使用 StructuredTaskScope.adapter 处理 Reactor 异常

为了解决上述 Reactor 异常逃逸的问题，我们可以使用 StructuredTaskScope.adapter 将 Reactor 的执行环境适配到 StructuredTaskScope 中。

修改后的代码如下：

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class StructuredConcurrencyReactorAdapterExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
        try (var scope = new StructuredTaskScope.ShutdownOnFailure("ReactorScope", executorService)) {
            Future<String> reactorFuture = scope.fork(() -> {
                Mono<String> mono = Mono.just("Initial Value")
                    .map(s -> {
                        if (s.equals("Initial Value")) {
                            throw new RuntimeException("Simulated Error in Reactor");
                        }
                        return s.toUpperCase();
                    });

                // 使用 errorConsumer 捕获异常
                mono.subscribe(
                    System.out::println,
                    e -> {
                        System.err.println("Error in Reactor: " + e.getMessage());
                        throw new RuntimeException("Reactor Error", e); // 重新抛出异常，让StructuredTaskScope感知
                    }
                );
                return "Reactor Task Completed";
            });

            scope.join();
            try {
                scope.throwIfFailed();
            } catch (Exception e) {
                System.err.println("StructuredTaskScope caught: " + e.getMessage());
            }

            System.out.println(reactorFuture.resultNow());
        } catch (Exception e) {
            System.err.println("Outer Error: " + e.getMessage());
        } finally {
            executorService.shutdown();
        }
    }
}

在这个修改后的例子中，我们首先创建了一个虚拟线程池 executorService，然后使用 StructuredTaskScope.ShutdownOnFailure("ReactorScope", executorService) 创建了一个适配的 StructuredTaskScope。关键在于 subscribe 方法中，添加了 errorConsumer, 并且在 errorConsumer 中重新抛出了 RuntimeException, 这样才能被 StructuredTaskScope 感知到。

此外，executorService 用完要 shutdown。

Hooks.onOperatorError 与异常处理

Reactor 提供了 Hooks.onOperatorError 钩子，允许我们全局地处理操作符中的异常。这个钩子可以用于记录日志、发送警报等。

Hooks.onOperatorError 的一个典型用法如下：

import reactor.core.publisher.Hooks;
import reactor.core.publisher.Mono;

public class HooksOnOperatorErrorExample {

    public static void main(String[] args) {
        Hooks.onOperatorError((e, context) -> {
            System.err.println("Global Error Handler: " + e.getMessage() + ", Context: " + context);
            return e; // 返回原始异常
        });

        Mono.just("Value")
            .map(s -> {
                throw new RuntimeException("Simulated Error");
            })
            .subscribe(
                System.out::println,
                Throwable::printStackTrace
            );
    }
}

在这个例子中，Hooks.onOperatorError 注册了一个全局的异常处理器。当 map() 操作符抛出异常时，这个异常处理器会被调用，打印错误信息和上下文信息。

Hooks.onOperatorError 与 StructuredTaskScope.adapter 的关系在于，Hooks.onOperatorError 可以全局地处理 Reactor 操作符中的异常，而 StructuredTaskScope.adapter 则可以将这些异常传递到 StructuredTaskScope 中，从而保证结构化并发的结构性。

然而，需要注意的是，Hooks.onOperatorError 是一个全局的钩子，会影响到整个 Reactor 应用程序。因此，在使用 Hooks.onOperatorError 时，需要谨慎考虑其影响范围，避免对其他模块造成不必要的影响。

最佳实践

在使用结构化并发与 Reactor 结合时，以下是一些最佳实践：

始终提供 errorConsumer: 在 Mono.subscribe() 方法中，始终提供 errorConsumer 来捕获异常。这可以避免异常逃逸到结构化并发的范围之外。
使用 StructuredTaskScope.adapter: 如果需要在结构化并发的范围内使用 Reactor，可以使用 StructuredTaskScope.adapter 将 Reactor 的执行环境适配到 StructuredTaskScope 中。
谨慎使用 Hooks.onOperatorError: Hooks.onOperatorError 是一个全局的钩子，在使用时需要谨慎考虑其影响范围。
重新抛出异常: 在 errorConsumer 中捕获异常后，将其重新抛出，以便 StructuredTaskScope 能够检测到任务失败。
使用虚拟线程池: 使用 Executors.newVirtualThreadPerTaskExecutor() 创建虚拟线程池，以提高并发程序的性能。

实践	说明
提供 `errorConsumer`	确保 `Mono.subscribe()` 方法包含 `errorConsumer`，以捕获 Reactor 序列中发生的异常。
使用 `StructuredTaskScope.adapter`	当在结构化并发范围内使用 Reactor 时，使用 `StructuredTaskScope.adapter` 将 Reactor 的执行环境适配到 `StructuredTaskScope` 中，以便正确处理异常。
谨慎使用 `Hooks.onOperatorError`	由于 `Hooks.onOperatorError` 是一个全局钩子，使用时需仔细评估其对整个 Reactor 应用的影响。
重新抛出异常	在 `errorConsumer` 捕获异常后，重新抛出异常，以便 `StructuredTaskScope` 可以检测到任务失败。
使用虚拟线程池	使用 `Executors.newVirtualThreadPerTaskExecutor()` 创建虚拟线程池，以提高并发程序的性能。

案例分析

我们来看一个更复杂的案例，演示如何使用结构化并发和 Reactor 处理多个并发任务，并正确处理异常：

import reactor.core.publisher.Mono;
import java.time.Duration;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

public class ComplexStructuredConcurrencyReactorExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executorService = Executors.newVirtualThreadPerTaskExecutor();
        try (var scope = new StructuredTaskScope.ShutdownOnFailure("ComplexReactorScope", executorService)) {

            Future<String> userFuture = scope.fork(() -> {
                return Mono.fromCallable(() -> fetchUserData())
                    .timeout(Duration.ofSeconds(2)) // 添加超时
                    .subscribeOn(reactor.core.scheduler.Schedulers.fromExecutorService(executorService)) // 在虚拟线程池中执行
                    .onErrorResume(e -> {
                        System.err.println("Error fetching user data: " + e.getMessage());
                        return Mono.just("Default User"); // 提供默认值
                    })
                    .block(); // 阻塞直到完成，或者在超时后返回
            });

            Future<Integer> orderFuture = scope.fork(() -> {
                return Mono.fromCallable(() -> fetchOrderCount())
                    .timeout(Duration.ofSeconds(1))  //添加超时
                    .subscribeOn(reactor.core.scheduler.Schedulers.fromExecutorService(executorService)) // 在虚拟线程池中执行
                    .onErrorResume(e -> {
                        System.err.println("Error fetching order count: " + e.getMessage());
                        return Mono.just(-1); // 提供默认值
                    })
                    .block(); // 阻塞直到完成，或者在超时后返回
            });

            scope.join();
            scope.throwIfFailed();

            String user = userFuture.resultNow();
            Integer orderCount = orderFuture.resultNow();

            System.out.println("User: " + user + ", Order Count: " + orderCount);

        } catch (Exception e) {
            System.err.println("Outer Error: " + e.getMessage());
        } finally {
            executorService.shutdown();
        }
    }

    static String fetchUserData() {
        try {
            Thread.sleep(1500);
            return "Real User Data";
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }

    static Integer fetchOrderCount() {
        try {
            Thread.sleep(500);
            // 模拟错误
            throw new RuntimeException("Failed to fetch order count");

            //return 123;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        }
    }
}

在这个案例中，我们使用了两个 Mono 序列来分别获取用户数据和订单数量。为了提高程序的健壮性，我们添加了超时机制，并使用了 onErrorResume 方法来处理异常，提供默认值。同时, 关键点在于使用了 subscribeOn , 将 Mono 的执行放在 executorService 中。

总结一下要点

总而言之，Java 21 的结构化并发特性为并发编程提供了更强的结构性和可靠性。结合 Project Reactor 使用时，需要特别注意异常处理，避免异常逃逸。StructuredTaskScope.adapter 可以将 Reactor 的执行环境适配到 StructuredTaskScope 中，从而保证结构化并发的结构性。同时，合理使用 Hooks.onOperatorError 可以全局地处理 Reactor 操作符中的异常，但需要谨慎考虑其影响范围。遵循最佳实践，可以编写出更加健壮和可靠的并发程序。

核心概念回顾

结构化并发通过 StructuredTaskScope 管理并发任务的生命周期。
Reactor 的 Mono.subscribe() 需要妥善处理异常，避免逃逸。
StructuredTaskScope.adapter 用于将 Reactor 执行环境适配到结构化并发中。
Hooks.onOperatorError 提供全局的 Reactor 异常处理机制。