MK
摩柯社区 - 一个极简的技术知识社区
AI 面试

Java反应式编程中的异常处理

2022-05-247.7k 阅读

Java 反应式编程简介

在现代软件开发中，反应式编程因其能够高效处理异步和事件驱动的场景而备受青睐。Java 作为一门广泛应用的编程语言，也积极拥抱反应式编程范式。Java 的反应式编程主要基于 Reactor 库，它提供了 Flux 和 Mono 这两个关键类型来表示异步数据流。

Flux 代表一个可以发出 0 到多个元素并且可以结束于成功完成或错误的异步序列。例如，以下代码创建了一个 Flux，它会发出 1 到 3 这三个整数：

Flux<Integer> flux = Flux.just(1, 2, 3);

Mono 则代表一个异步操作，它最多只会产生一个元素，或者结束于成功完成或错误。比如创建一个 Mono，它返回一个固定的字符串：

Mono<String> mono = Mono.just("Hello, Reactive World!");

反应式编程通过操作符链来处理数据流，这些操作符可以对数据进行转换、过滤、组合等操作，使得代码更加简洁和可读，同时也能有效利用异步特性提高系统性能。

异常在反应式编程中的挑战

在传统的同步编程中，异常处理相对直接，通常使用 try - catch 块来捕获和处理异常。例如：

try {
    int result = 10 / 0;
} catch (ArithmeticException e) {
    System.out.println("Caught an arithmetic exception: " + e.getMessage());
}

然而，在反应式编程中，情况变得更为复杂。由于数据流是异步的，异常不能像同步代码那样简单地通过 try - catch 块捕获。当一个操作符链中的某个操作抛出异常时，这个异常需要在整个反应式序列的上下文中进行处理，而不是在抛出异常的局部代码块中。

例如，考虑以下 Flux 操作链：

Flux.just(1, 2, 3)
   .map(i -> 10 / i)
   .subscribe(System.out::println);

在这个例子中，如果 map 操作符中的 10 / i 计算出现除零异常，传统的 try - catch 块无法直接捕获这个异常，因为整个操作是异步的。这就需要使用反应式编程特有的异常处理机制。

基本的异常处理操作符

onErrorReturn

onErrorReturn 操作符用于在数据流中发生异常时返回一个默认值。当 Flux 或 Mono 在处理过程中抛出异常时，onErrorReturn 会捕获这个异常，并返回预先设定的默认值，而不是让异常传播导致序列终止。

以下是一个示例，演示如何使用 onErrorReturn：

Flux.just(1, 0, 3)
   .map(i -> 10 / i)
   .onErrorReturn(-1)
   .subscribe(System.out::println);

在上述代码中，当 map 操作符处理到 0 时会抛出 ArithmeticException，onErrorReturn 捕获到这个异常并返回 -1。最终输出结果为 10，-1，3。

onErrorResume

onErrorResume 比 onErrorReturn 更灵活，它允许在发生异常时切换到另一个 Flux 或 Mono 序列。当原始序列发生异常时，onErrorResume 会根据异常类型选择一个备用的序列继续进行处理。

示例代码如下：

Flux.just(1, 0, 3)
   .map(i -> 10 / i)
   .onErrorResume(ArithmeticException.class, e -> Flux.just(-1, -2, -3))
   .subscribe(System.out::println);

在这个例子中，当遇到除零异常时，onErrorResume 会切换到 Flux.just(-1, -2, -3) 这个备用序列，最终输出 -1，-2，-3。

doOnError

doOnError 操作符并不改变数据流，它主要用于在发生异常时执行一些副作用操作，比如记录日志。它只是在异常发生时被调用，异常仍然会继续传播，除非被其他异常处理操作符捕获。

示例如下：

Flux.just(1, 0, 3)
   .map(i -> 10 / i)
   .doOnError(ArithmeticException.class, e -> System.out.println("Caught an arithmetic exception: " + e.getMessage()))
   .subscribe(System.out::println);

在这个代码中，当除零异常发生时，doOnError 会打印异常信息，但异常仍然会导致序列终止，最终只会输出 10。

全局异常处理

在实际应用中，特别是在大型项目中，对整个反应式应用进行全局异常处理是非常必要的。这可以确保所有未处理的异常都能得到一致的处理，避免应用因为未捕获的异常而崩溃。

在 Spring Boot 应用中，可以通过实现 ErrorWebExceptionHandler 接口来实现全局异常处理。以下是一个简单的示例：

import org.springframework.boot.web.reactive.error.ErrorWebExceptionHandler;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.annotation.Order;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.http.server.reactive.ServerHttpResponse;
import org.springframework.web.server.ResponseStatusException;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;

@Configuration
@Order(-2)
public class GlobalErrorHandler implements ErrorWebExceptionHandler {

    @Override
    public Mono<Void> handle(ServerWebExchange exchange, Throwable ex) {
        ServerHttpRequest request = exchange.getRequest();
        ServerHttpResponse response = exchange.getResponse();

        if (ex instanceof ResponseStatusException) {
            ResponseStatusException statusException = (ResponseStatusException) ex;
            response.setStatusCode(statusException.getStatus());
        } else {
            response.setStatusCode(HttpStatus.INTERNAL_SERVER_ERROR);
        }

        response.getHeaders().setContentType(MediaType.APPLICATION_JSON);
        return response.writeWith(Mono.fromSupplier(() -> {
            // 这里可以构建更详细的错误响应体
            return response.bufferFactory().wrap(("{\"error\":\"" + ex.getMessage() + "\"}").getBytes());
        }));
    }
}

这个全局异常处理器会捕获所有未处理的异常，并根据异常类型设置合适的 HTTP 状态码，同时返回一个包含错误信息的 JSON 响应。

异常处理与背压

背压（Backpressure）是反应式编程中处理数据流速度不匹配的机制。在存在背压的情况下，异常处理需要额外小心。当一个慢速的订阅者无法跟上快速产生数据的发布者时，背压策略会起作用。

例如，假设我们有一个快速产生数据的 Flux 和一个慢速处理数据的订阅者：

Flux.range(1, 10000)
   .publishOn(Schedulers.parallel())
   .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnNext(Integer value) {
            try {
                Thread.sleep(100);
                System.out.println("Received: " + value);
            } catch (InterruptedException e) {
                cancel();
            }
        }
    });

在这个例子中，Flux.range(1, 10000) 会快速产生 10000 个整数，而订阅者通过 Thread.sleep(100) 模拟慢速处理。如果没有背压处理，可能会导致内存溢出等问题。

当发生异常时，背压策略可能会影响异常的传播和处理。例如，如果在背压处理过程中发生异常，需要确保异常能够正确地被捕获和处理，而不会导致系统不稳定。可以结合前面提到的异常处理操作符，如 onErrorReturn、onErrorResume 等来处理背压过程中的异常。

异常处理与重试机制

在反应式编程中，重试机制是处理临时故障的常用手段。例如，当网络请求失败或者数据库连接暂时不可用时，我们可能希望重试操作，而不是立即返回错误。

Reactor 提供了 retry 操作符来实现重试功能。retry 操作符会在发生异常时重新订阅源序列，尝试再次执行操作。

以下是一个简单的示例，模拟网络请求失败并重试：

Mono<String> networkRequest = Mono.fromCallable(() -> {
    if (Math.random() < 0.5) {
        throw new RuntimeException("Network request failed");
    }
    return "Success response";
});

networkRequest
   .retry(3)
   .subscribe(System.out::println, System.err::println);

在这个例子中，networkRequest 模拟一个网络请求，有 50% 的概率失败。retry(3) 表示如果请求失败，最多重试 3 次。如果 3 次重试后仍然失败，异常会被传播并由订阅者的错误处理逻辑处理。

组合多个异常处理操作符

在实际应用中，往往需要组合多个异常处理操作符来满足复杂的业务需求。例如，我们可能先希望对特定类型的异常进行重试，然后在重试一定次数后，如果仍然失败，返回一个默认值。

以下代码展示了这种组合：

Flux.just(1, 0, 3)
   .map(i -> 10 / i)
   .retryWhen(Retry.backoff(3, Duration.ofSeconds(1))
       .filter(throwable -> throwable instanceof ArithmeticException))
   .onErrorReturn(-1)
   .subscribe(System.out::println);

在这个例子中，retryWhen 操作符会对 ArithmeticException 进行重试，最多重试 3 次，每次重试间隔 1 秒。如果 3 次重试后仍然失败，onErrorReturn 操作符会返回 -1。

自定义异常处理

除了使用 Reactor 提供的标准异常处理操作符，我们还可以根据业务需求自定义异常处理逻辑。这可以通过创建自定义的操作符来实现。

例如，假设我们希望在发生特定异常时，不仅返回一个默认值，还记录详细的异常日志到文件中。我们可以创建一个自定义的操作符：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.MonoSink;
import reactor.core.publisher.Operators;
import reactor.util.context.Context;

import java.io.FileWriter;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;

public class CustomErrorHandling {

    private static final Logger LOGGER = Logger.getLogger(CustomErrorHandling.class.getName());

    public static <T> Mono<T> onErrorLogAndReturn(T defaultValue, Class<? extends Throwable> exceptionClass) {
        return Operators.lift((Mono<T> mono) -> mono.handle((value, sink) -> {
            try {
                sink.next(value);
                sink.complete();
            } catch (Throwable t) {
                if (exceptionClass.isInstance(t)) {
                    logException(t);
                    sink.success(defaultValue);
                } else {
                    sink.error(t);
                }
            }
        }));
    }

    private static void logException(Throwable t) {
        try (FileWriter writer = new FileWriter("error.log", true)) {
            t.printStackTrace(new java.io.PrintWriter(writer));
        } catch (IOException e) {
            LOGGER.log(Level.SEVERE, "Error writing to error log", e);
        }
    }
}

然后可以在反应式代码中使用这个自定义操作符：

Flux.just(1, 0, 3)
   .map(i -> 10 / i)
   .transform(CustomErrorHandling.onErrorLogAndReturn(-1, ArithmeticException.class))
   .subscribe(System.out::println);

在这个例子中，当发生 ArithmeticException 时，异常会被记录到 error.log 文件中，并且返回 -1。

总结异常处理在不同场景下的最佳实践

  1. 简单默认值返回：如果只需要在异常发生时返回一个简单的默认值，onErrorReturn 是最佳选择。例如，在一个获取用户信息的操作中，如果由于网络故障无法获取到用户信息，可以返回一个默认的用户对象。
  2. 备用序列切换：当需要在异常发生时切换到另一个备用的数据流时，onErrorResume 更为合适。比如在从主数据库读取数据失败时，切换到从数据库读取数据。
  3. 日志记录：对于记录异常日志，doOnError 是很好的选择。它不会改变数据流，只是在异常发生时执行记录日志的操作。
  4. 全局处理：在大型应用中，实现全局异常处理，如通过 ErrorWebExceptionHandler 来确保所有未处理的异常都能得到统一处理，提高应用的稳定性。
  5. 重试机制：对于处理临时故障，retry 操作符非常有用。结合 retryWhen 可以根据异常类型和重试策略进行灵活的重试处理。
  6. 自定义处理：当标准的异常处理操作符无法满足业务需求时，可以通过自定义操作符来实现特定的异常处理逻辑，如记录详细异常日志并返回自定义默认值。

通过合理运用这些异常处理机制，可以使 Java 反应式编程更加健壮和可靠，能够有效应对各种异常情况，确保应用的稳定运行。