关于响应式流Reactive Streams-牛翰网

响应式流处理引入了类似于观察者模式（Observer Pattern）的异步、非阻塞、事件驱动的编程范式，允许数据作为连续的流进行处理。它可以处理异步数据流，并支持 back-pressure（反压），这意味着消费者可以以它们能够处理的速度来消费数据。

1.Project Reactor

Project Reactor 是一个实现了响应式流规范的 Java 库，提供了丰富的 API 用于编程处理异步数据流。任何时候数据流都可以被视为一个系列的事件，这在概念上与 Reactor 模式类似。
Project Reactor为 Java 提供了异步和事件驱动的编程模型。这个库主要提供了两种核心的响应式类型：Mono 和 Flux，以及管理异步任务执行的 Schedulers。

Mono

Mono 代表一个单一的异步计算值，可以发出0或1个元素。它是处理单一异步操作的理想选择，例如异步的HTTP请求响应、单个数据库查询等。

使用示例：

import reactor.core.publisher.Mono;

public class MonoExample {
    public static void main(String[] args) {
        Mono<String> mono = Mono.just("Hello World");
        mono.subscribe(System.out::println); // 输出 Hello World
    }
}

在这个例子中，Mono.just 创建了一个包含单一值的 Mono 实例，并通过 subscribe 方法输出这个值。

优点：

简洁的 API，适合处理单一值或单一事件。
支持简单和复杂的异步转换操作。
减少资源消耗（相比返回整个集合的服务）。

缺点：

对于初学者而言，响应式编程的学习曲线较陡峭。
错误处理和调试可能比较复杂。

Flux

Flux 代表一个异步序列，可以发出0到N个元素。它适用于处理多值响应，如数据流、事件流或其他类型的元素集合。

使用示例：

import reactor.core.publisher.Flux;

public class FluxExample {
    public static void main(String[] args) {
        Flux<Integer> flux = Flux.just(1, 2, 3, 4);
        flux.subscribe(System.out::println); // 输出 1, 2, 3, 4
    }
}

在这里，Flux.just 生成一个包含多个整数的 Flux 实例，并通过 subscribe 输出每个整数。

优点：

灵活地处理多个数据。
支持多种反压策略来应对数据流速率的控制。
强大的操作符集合，支持复杂的数据流转换和合并操作。

缺点：

资源消耗可以随着流中元素数量的增加而增加。
需要更多的控制和错误处理策略。

Schedulers

Schedulers 是 Reactor 提供的一个组件，用于控制异步任务在不同的工作线程或线程池上执行。它使得开发者可以灵活地控制执行环境，优化性能，尤其适合于处理IO密集或CPU密集的任务。

使用示例：

Mono.fromCallable(() -> {
    return "Some IO operation result";
})
    .subscribeOn(Schedulers.boundedElastic())
    .subscribe(System.out::println);

在这个例子中，使用 Schedulers.boundedElastic() 确保在一个弹性的、旨在为阻塞任务优化的线程池中执行 I/O 操作。

优点：

灵活的线程管理，能够根据任务特性优化线程的使用。
可以避免阻塞主线程，提高应用的响应性和吞吐量。

缺点：

管理不当可能导致资源过度使用，如创建过多的线程。
可能增加应用的复杂性。

每种组件都有其适用场景和限制。选择合适的类型和调度器，可以帮助开发高效、响应快速的应用。

2. Schedulers.boundedElastic()

在响应式编程和特别是使用诸如 Project Reactor 这样的库时，确保系统的鲁棒性和避免像内存耗尽（Out of Memory, OOM）这种问题非常重要。在响应式流中使用 subscribeOn(Schedulers.boundedElastic()) 联合 onErrorResume 方法可以在一定程度上帮助管理资源并处理错误，从而避免系统因异常情况如OOM而崩溃。让我们逐一看看这些组件怎样工作，并解释其对预防OOM的潜在影响。

Schedulers.boundedElastic()
在 Project Reactor 中，Schedulers.boundedElastic() 创建了一个可以根据需要弹性扩展的调度器，但有一定的限制，用于避免无限制地创建线程。boundedElastic调度器主要用于执行阻塞操作或长时间运行的任务，而不会消耗过多的系统资源（例如线程）。线程数的上限默认为CPU核数的10倍，任务队列的大小限制为10,000。当线程和队列达到上限时，进一步的任务提交会被延迟，直到有空闲资源。

这种调度器的使用尤其适合处理那些不适合在响应式流的主线程中执行的阻塞I/O操作，因为这些操作会阻塞处理流程，减慢系统反应速度。

onErrorResume
onErrorResume 是一个错误处理操作符，它允许你在发生错误（如抛出异常）时提供一个备用的数据流。这意味着，如果原始流中出现了错误，比如因为OutOfMemoryError而失败，你可以决定终止出错的流并继续执行，恢复成一个预设的备用流，或者简单地清理资源并进行适当的日志记录。

避免OOM的潜在影响
使用 subscribeOn(Schedulers.boundedElastic()) 允许长时间运行的或阻塞的任务在单独的线程中执行，这样主流程就不会被阻塞。这种方式可减少主系统线程的负载，从而降低因资源耗尽导致OOM的风险。

通过 onErrorResume 提供的错误处理机制，当流中的某个部分出现问题（例如，因为无法处理的大量数据而导致的OOM），程序可以捕获这个异常并优雅地处理，比如释放资源、调整内存使用、或者切换到更为节省内存的操作，而不是让整个应用崩溃。

public Mono<R> receiveJson(@RequestBody String jsonData) {
        return Mono.fromCallable(() -> {
            senderService.senderMsg(jsonData, TUYA_TOPIC);
            return OK;
        }).subscribeOn(Schedulers.boundedElastic()).onErrorResume(e -> {
            log.error("Error occurred: {}", e.getMessage());
            return Mono.just(new R(-1, "An error occurred: " + e.getMessage()));
        });
    }

综合使用这两个方法不仅增强了程序的健壮性，而且提供了更细粒度的资源控制和错误处理选项，有助于在面对资源限制和不可预测错误时更好地维护应用的稳定运行。然而，它们并不能直接防止OOM异常的发生，正确的数据管理和流控制策略才是根本解决方式。这包括合理的反压策略、数据分批处理、合理的内存和资源分配等措施。

3. 反压

反压（Back-Pressure）的概念

在响应式流处理中，反压（Back-Pressure）是一种重要的机制，用于处理数据生产者（Publisher）和消费者（Subscriber）之间速率不一致的问题。当生产者产生数据的速度超过消费者处理数据的速度时，未处理的数据可能会在内存中积压，导致资源耗尽，甚至应用程序崩溃。反压机制允许消费者根据自己的处理能力，控制生产者的数据生产速度，从而有效避免这一问题。

反压的工作原理

在响应式编程标准（如Reactive Streams）中，当消费者准备好接收新的数据项时，它会向生产者发送一个请求信号，指明可以发送多少数据项。这样，生产者在发送数据前会等待这个请求信号，从而确保不会发送消费者处理不了的数据量。

应用场景

实时数据流处理：例如，处理来自传感器或用户界面事件的数据流，其中数据产生速度可能非常快。
资源有限的环境：在内存或处理能力有限的设备上，如移动设备或嵌入式系统，反压机制可以防止系统过载。
高负载系统：在服务器处理大量并发请求或大数据处理任务时，反压机制帮助系统平衡负载，提高稳定性。

如何使用反压

在Java中，通过发布订阅框架（如Reactor或RxJava）可以很容易地实现反压。

使用 Project Reactor 实现反压：

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

public class BackPressureExample {
    public static void main(String[] args) {
        Flux.range(1, 100) // 生产一个1到100的数列
            .onBackpressureDrop() // 当Backpressure时，丢弃无法处理的数据
            .publishOn(Schedulers.parallel(), 1) // 指定使用并行调度器，并设置最大请求量为1
            .subscribe(
                data -> {
                    System.out.println("Received: " + data);
                    try {
                        Thread.sleep(100); // 模拟慢消费者
                    } catch (InterruptedException e) {
                        Thread.currentThread().interrupt();
                    }
                },
                err -> System.err.println("Error: " + err),
                () -> System.out.println("Completed")
            );
    }
}

在这个例子中，.onBackpressureDrop()操作符用于指定如何处理溢出数据。当消费者处理速度跟不上生产者时，超出的数据将被丢弃。

当然也有其他的反压策略：

onBackpressureBuffer()
这个策略会缓存所有来自上游的数据。如果下游处理不过来，数据将在内存中缓存。这可以防止数据丢失，但可能会占用大量内存，如果缓存过多，仍然有可能导致内存溢出（OutOfMemoryError）。
```
Flux.range(1, 100)
	.onBackpressureBuffer()
	.publishOn(Schedulers.parallel(), 1)
	.subscribe(...);
```
onBackpressureLatest()
此策略保留最新的数据，丢弃旧的数据。当下游处理速度跟不上时，这种方式只会保留最新的一条数据，老的数据会被覆盖。这比 onBackpressureDrop() 策略提供了更多的数据保证，至少最新的数据是可用的。
```
Flux.range(1, 100)
	.onBackpressureLatest()
	.publishOn(Schedulers.parallel(), 1)
	.subscribe(...);
```
onBackpressureError()
这种策略会立即报告一个错误，一旦下游不能及时处理上游的数据。这适用于数据完整性非常重要的场合，任何数据丢失都是不可接受的。
```
Flux.range(1, 100)
	.onBackpressureError()
	.publishOn(Schedulers.parallel(), 1)
	.subscribe(...);
```

动态调节请求量
可以根据消费者的实际处理能力动态调调节请求的数据量。这通常需要手动控制请求，而不是使用 publishOn() 方式自动请求。

Flux.range(1, 100)
	.subscribe(new BaseSubscriber<Integer>() {
		@Override
		protected void hookOnSubscribe(Subscription subscription) {
			request(1); // Initial request
		}

		@Override
		protected void hookOnNext(Integer value) {
			process(value);
			request(1); // Request one more after processing
		}
	});

使用自定义策略
通过自定义 Subscriber 和使用 reactor.core.publisher.Flux.create 或 generate 方法，你可以实现完全自定义的背压管理策略。

选择哪种策略应根据应用的具体需求、数据重要性以及系统资源限制来决定。如果对数据完整性要求较高，则应避免使用丢弃数据的策略。如果系统资源有限，可能需要更多考虑动态调节或缓存策略。