JAVA Reactor 流数据丢失？正确使用 onErrorResume 与订阅策略

大家好，今天我们来深入探讨在使用 Java Reactor 处理流数据时可能遇到的数据丢失问题，以及如何通过正确使用 onErrorResume 和订阅策略来避免这些问题。Reactor 是一个响应式编程库，它允许我们以声明式的方式处理异步数据流。然而，不当的使用会导致数据丢失，尤其是在处理错误时。

1. 数据丢失的常见场景

在使用 Reactor 处理数据流时，数据丢失通常发生在以下几个场景：

未处理的异常： 如果在数据流处理过程中抛出未捕获的异常，整个流可能会被终止，导致后续的数据无法被处理。
不正确的错误处理： 使用错误的错误处理方式，例如忽略错误或在错误处理逻辑中引入新的错误，可能会导致数据丢失或流的提前终止。
背压策略不当： 在生产者速度快于消费者速度的情况下，如果背压策略配置不当，可能会导致数据被丢弃。

2. onErrorResume 的作用与陷阱

onErrorResume 是 Reactor 中一个重要的错误处理操作符。它的作用是在流遇到错误时，切换到另一个预定义的 Publisher。这允许我们在发生错误时提供备选数据或执行恢复逻辑，而不是直接终止流。

2.1 onErrorResume 的基本用法

onErrorResume 接收一个 Function<Throwable, ? extends Publisher<? extends T>> 作为参数。这个函数接收一个 Throwable 对象（表示发生的错误），并返回一个 Publisher，该 Publisher 将被用来替代发生错误的原始 Publisher。

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class OnErrorResumeExample {

    public static void main(String[] args) {
        Flux<Integer> flux = Flux.just(1, 2, 0, 4, 5)
                .map(i -> 10 / i) // 除以 0 会抛出 ArithmeticException
                .onErrorResume(e -> {
                    System.err.println("发生错误: " + e.getMessage());
                    return Flux.just(-1, -2); // 提供备选数据
                });

        flux.subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("完成")
        );
    }
}

在这个例子中，当 flux 遇到 ArithmeticException（由于除以 0）时，onErrorResume 会被触发。它将错误信息打印到控制台，并返回一个新的 Flux，该 Flux 发出 -1 和 -2。最终，输出将是：

10
5
发生错误: / by zero
-1
-2
完成

2.2 onErrorResume 的常见陷阱

虽然 onErrorResume 是一个强大的工具，但如果不小心使用，也可能导致数据丢失或不正确的行为。

吞噬异常： 最常见的错误是仅仅记录错误而不采取任何恢复措施，或者返回一个空的 Publisher。这会导致数据流静默地丢失数据。

Flux<Integer> flux = Flux.just(1, 2, 0, 4, 5)
        .map(i -> 10 / i)
        .onErrorResume(e -> {
            System.err.println("发生错误: " + e.getMessage());
            return Flux.empty(); // 错误！ 吞噬了异常并丢失了数据
        });

flux.subscribe(
        System.out::println,
        System.err::println,
        () -> System.out.println("完成")
);

在这个例子中，当发生 ArithmeticException 时，错误信息会被打印，但 Flux.empty() 会被返回，导致后续的数据（4 和 5）被丢失。输出将是：

10
5
发生错误: / by zero
完成

引入新的异常： 在 onErrorResume 的处理逻辑中，如果抛出新的异常，可能会导致原始的错误被掩盖，并且新的异常如果没有被处理，仍然会导致流的终止。

Flux<Integer> flux = Flux.just(1, 2, 0, 4, 5)
        .map(i -> 10 / i)
        .onErrorResume(e -> {
            System.err.println("发生错误: " + e.getMessage());
            throw new RuntimeException("onErrorResume 中发生新的错误"); // 错误！引入新的异常
        });

flux.subscribe(
        System.out::println,
        System.err::println,
        () -> System.out.println("完成")
);

在这个例子中，当发生 ArithmeticException 时，onErrorResume 会抛出一个新的 RuntimeException。这个新的异常会被传递给 subscribe 的错误处理方法，导致原始的 ArithmeticException 被掩盖。输出将是：

10
5
发生错误: / by zero
java.lang.RuntimeException: onErrorResume 中发生新的错误

过度使用 onErrorResume： 不应该将 onErrorResume 作为默认的错误处理方式。在某些情况下，终止流并通知调用者错误可能更合适。过度使用 onErrorResume 会隐藏潜在的问题，并使调试变得更加困难。

2.3 正确使用 onErrorResume 的最佳实践

提供有意义的备选数据： 确保 onErrorResume 返回的 Publisher 提供有意义的备选数据。备选数据应该能够代替原始数据，并且不会对后续的处理逻辑产生负面影响。
记录错误并采取适当的恢复措施： 在 onErrorResume 中，应该记录错误信息，并采取适当的恢复措施。恢复措施可能包括重试操作、使用缓存数据或通知其他系统。
避免引入新的异常： 尽量避免在 onErrorResume 的处理逻辑中抛出新的异常。如果必须抛出新的异常，应该确保新的异常能够被正确处理。
谨慎使用 onErrorResume： 只有在能够提供有意义的恢复逻辑时才应该使用 onErrorResume。在其他情况下，应该考虑使用其他错误处理操作符，例如 onErrorReturn、onErrorMap 或 onErrorContinue。

3. 订阅策略的重要性

订阅策略定义了当 Publisher 发出数据时，Subscriber 如何接收和处理数据。正确的订阅策略对于防止数据丢失至关重要，尤其是在处理高吞吐量的数据流时。

3.1 背压的概念

背压是一种机制，允许消费者 (Subscriber) 向生产者 (Publisher) 发出信号，表明它能够处理的数据量。这可以防止生产者以超过消费者处理能力的速度发送数据，从而导致数据丢失或性能问题。

3.2 Reactor 的背压策略

Reactor 提供了多种背压策略，可以根据不同的应用场景进行选择：

策略	描述	适用场景
`BUFFER`	将所有未处理的数据缓冲起来。如果生产者速度远大于消费者，可能会导致内存溢出。	适用于生产者速度和消费者速度相差不大，并且有足够的内存来缓冲数据的情况。
`DROP`	当消费者无法处理数据时，直接丢弃新的数据。	适用于对数据完整性要求不高，并且希望保持系统响应速度的情况。例如，实时监控系统可以丢弃一些不重要的数据。
`LATEST`	只保留最新的数据，丢弃旧的数据。	适用于只关心最新数据的情况。例如，股票行情系统只关心最新的价格。
`ERROR`	当消费者无法处理数据时，发出一个错误信号。	适用于需要严格保证数据完整性的情况。例如，金融交易系统必须保证所有交易数据都被处理。
`IGNORE`	忽略背压信号，生产者继续以最大速度发送数据。这可能会导致数据丢失或性能问题。	除非非常清楚自己在做什么，否则不应该使用这种策略。
`UNBOUNDED`	使用 `Flux.create` 或 `Flux.generate` 创建 `Flux` 时，默认使用 `UNBOUNDED` 模式，意味着会尽可能快的生产数据，不考虑下游消费者的处理能力。除非使用 `onRequest` 手动控制生产速度，否则可能会导致 `OutOfMemoryError`。	同 `IGNORE`，除非使用 `onRequest` 手动控制生产速度，否则不应该使用这种策略。

3.3 如何选择合适的背压策略

选择合适的背压策略取决于具体的应用场景。以下是一些选择策略的建议：

数据完整性： 如果数据完整性非常重要，应该选择 ERROR 策略。
系统响应速度： 如果希望保持系统响应速度，可以考虑使用 DROP 或 LATEST 策略。
内存限制： 如果内存有限制，应该避免使用 BUFFER 策略。
生产者和消费者速度差异： 如果生产者速度和消费者速度相差不大，可以使用 BUFFER 策略。
明确的需求： 如果明确知道只需要最新的数据，可以使用 LATEST 策略。

3.4 使用 Flux.create 和 Flux.generate 时的背压处理

当使用 Flux.create 或 Flux.generate 创建 Flux 时，需要手动处理背压。这两个方法允许我们自定义数据的生成逻辑，但也要求我们负责处理背压信号。

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

public class FluxCreateExample {

    public static void main(String[] args) {
        Flux<Integer> flux = Flux.create(sink -> {
            for (int i = 0; i < 10; i++) {
                if (sink.isCancelled()) {
                    return; // 如果 Subscriber 取消订阅，则停止发送数据
                }
                sink.next(i);
            }
            sink.complete();
        }, FluxSink.OverflowStrategy.BUFFER); // 或者使用 DROP, LATEST, ERROR 等策略

        flux.subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("完成")
        );
    }
}

在这个例子中，我们使用 Flux.create 创建一个 Flux。FluxSink 提供了 next、error 和 complete 方法，用于发送数据、发出错误信号和完成流。为了处理背压，我们检查 sink.isCancelled() 方法，如果 Subscriber 取消订阅，则停止发送数据。FluxSink.OverflowStrategy.BUFFER 指定了当 sink 的缓冲区已满时，应该采取的策略。

一个更细粒度的控制背压的例子是使用 onRequest：

import reactor.core.publisher.Flux;

public class FluxGenerateExample {

    public static void main(String[] args) {
        Flux<Integer> flux = Flux.generate(
                () -> 0, // 初始化状态
                (state, sink) -> {
                    if (state > 10) {
                        sink.complete();
                        return state;
                    }
                    sink.next(state);
                    return state + 1;
                },
                (state) -> System.out.println("清理状态: " + state)
        );

        flux.subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("完成")
        );
    }
}

这个例子没有显式地处理背压，因为 Flux.generate 默认是 unbounded 模式，并且同步地生成数据。但是如果生成数据的逻辑是异步的，那么就需要使用 onRequest 来控制数据生成的速度。

4. 结合 onErrorResume 和订阅策略

onErrorResume 和订阅策略是两个不同的概念，但它们可以结合起来使用，以提供更健壮的错误处理和数据流控制。

在 onErrorResume 中考虑背压： 当在 onErrorResume 中返回一个新的 Publisher 时，应该确保新的 Publisher 也能够处理背压。例如，如果原始的 Publisher 使用 BUFFER 策略，那么新的 Publisher 也应该使用类似的策略。
使用 onErrorResume 来处理背压错误： 当使用 ERROR 策略时，如果发生背压错误，onErrorResume 可以用来提供备选数据或执行恢复逻辑。

5. 一个完整的例子

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

import java.time.Duration;
import java.util.Random;

public class CompleteExample {

    public static void main(String[] args) throws InterruptedException {
        Random random = new Random();

        Flux<Integer> flux = Flux.interval(Duration.ofMillis(100))
                .map(tick -> random.nextInt(10))
                .map(i -> {
                    if (i == 0) {
                        throw new RuntimeException("除以 0 错误");
                    }
                    return 100 / i;
                })
                .onErrorResume(e -> {
                    System.err.println("发生错误: " + e.getMessage());
                    return Flux.just(-1); // 发生错误时，使用 -1 作为默认值
                })
                .onBackpressureBuffer(10) // 使用 BUFFER 策略，最多缓冲 10 个元素
                .publishOn(Schedulers.newSingle("consumer-thread")) // 切换到消费者线程
                .doOnNext(i -> {
                    try {
                        Thread.sleep(500); // 模拟消费者处理数据的延迟
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                });

        flux.subscribe(
                System.out::println,
                System.err::println,
                () -> System.out.println("完成")
        );

        Thread.sleep(5000); // 运行 5 秒钟
    }
}

在这个例子中，我们创建了一个 Flux，它每 100 毫秒生成一个随机数，然后将 100 除以这个随机数。如果随机数为 0，则会抛出一个 RuntimeException。onErrorResume 用来捕获这个异常，并使用 -1 作为默认值。onBackpressureBuffer(10) 使用 BUFFER 策略，最多缓冲 10 个元素。publishOn(Schedulers.newSingle("consumer-thread")) 将数据流切换到消费者线程，并模拟消费者处理数据的延迟。这个例子演示了如何结合 onErrorResume 和订阅策略来处理错误和控制数据流。

6. 总结：掌握错误处理和背压是关键

数据丢失是响应式编程中一个常见的问题，但可以通过正确使用 onErrorResume 和订阅策略来避免。onErrorResume 允许我们在发生错误时提供备选数据或执行恢复逻辑，而订阅策略允许我们控制数据流的速度，防止生产者以超过消费者处理能力的速度发送数据。掌握这两种技术对于编写健壮的 Reactor 应用至关重要。

JAVA Reactor 流数据丢失？正确使用 onErrorResume 与订阅策略

发表回复 取消回复

发表回复取消回复