重学SpringBoot3-Reactive-Streams规范

更多SpringBoot3内容请关注我的专栏：《SpringBoot3》
期待您的点赞👍收藏⭐评论✍

在这里插入图片描述

重学SpringBoot3-Reactive-Streams规范

1. 什么是 Reactive-Streams 规范？
2. Reactive-Streams 的核心组件
- 2.1 Publisher（发布者）
- 2.2 Subscriber（订阅者）
- 2.3 Subscription（订阅）
- 2.4 Processor（处理器）
3. 背压机制（Backpressure）
4. Reactive-Streams 与 Reactor
5. 为什么选择 Reactive-Streams？
6. 总结

随着现代系统对高吞吐量、低延迟和可扩展性需求的增加，响应式编程逐渐成为处理异步数据流的重要范式。上一篇文章介绍了 Reactor核心概念，而 Reactor 的基础就是 Reactive-Streams 规范，它定义了一套标准化的异步数据处理接口，用于在不同的响应式编程框架和库之间实现兼容性。

在这篇博客中，我们将详细介绍 Reactive-Streams 规范的核心概念和它在实际编程中的重要性。

1. 什么是 Reactive-Streams 规范？

Reactive-Streams 是由多家技术公司（包括 Lightbend、Netflix、Pivotal 等）联合发布的一套处理异步流式数据的标准。其核心目标是定义一个兼容的、非阻塞的背压（Backpressure）处理模型，帮助开发者处理高速数据流中可能产生的压迫问题。

Reactive-Streams 规范主要针对以下几个问题：

异步数据流的处理：以非阻塞方式处理数据，保证资源高效使用。
背压处理：当消费者的处理速度低于生产者时，合理管理数据流的流量，避免系统崩溃。
跨框架兼容性：在不同响应式框架（如 Reactor、RxJava 等）之间实现互操作。

2. Reactive-Streams 的核心组件

Reactive-Streams 规范定义了四个核心接口，分别为 Publisher、Subscriber、Subscription 和 Processor。这些接口共同构成了异步数据流的处理模型。

2.1 Publisher（发布者）

Publisher 负责发布数据，它是数据源的一部分，向订阅者（Subscriber）发送数据。Publisher 接口非常简单，定义了一个方法：

public interface Publisher<T> {void subscribe(Subscriber<? super T> subscriber);
}

通过 subscribe 方法，Publisher 可以向多个 Subscriber 注册，通知其数据流的到达。

2.2 Subscriber（订阅者）

Subscriber 是数据的消费者，接收 Publisher 发布的数据流。Subscriber 需要实现四个方法，分别处理不同的状态变化：

public interface Subscriber<T> {void onSubscribe(Subscription s);  // 初始化时调用void onNext(T t);                  // 当有新数据到达时调用void onError(Throwable t);         // 当发生错误时调用void onComplete();                 // 当数据流结束时调用
}

onSubscribe：接收到 Subscription 对象，订阅者可以通过它控制数据的请求和取消。
onNext：每当有数据发布时，Publisher 会调用该方法。
onError：如果发生错误，onError 会被调用，终止数据流。
onComplete：当所有数据发布完成时调用。

2.3 Subscription（订阅）

Subscription 是连接 Publisher 和 Subscriber 的纽带，它允许 Subscriber 控制数据流的数量。背压机制就依赖于 Subscription 进行数据流量控制：

public interface Subscription {void request(long n);   // 请求 n 个数据元素void cancel();          // 取消数据流
}

request：Subscriber 使用 request 方法向 Publisher 请求一定数量的数据，避免数据泛滥。
cancel：终止数据流，停止接收任何新的数据。

2.4 Processor（处理器）

Processor 是一种特殊的组件，它既是 Subscriber 也是 Publisher，充当中间处理器，允许在接收到数据后对其进行处理再发布给下游。

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {// 既能订阅数据，也能发布处理后的数据
}

3. 背压机制（Backpressure）

背压是 Reactive-Streams 规范中的关键概念。它用于处理生产者发送数据过快（正压），而消费者无法及时处理的情况。没有背压机制的系统很容易出现内存溢出或性能下降。

通过 Subscription 的 request(n) 方法，消费者可以根据自己的处理能力，向生产者请求合适数量的数据。如果消费者处理不过来，它可以在没有请求更多数据之前停止接收。

以下是一个简单的背压示例：

package com.coderjia.boot3webflux.controller;import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;/*** @author CoderJia* @create 2024/10/21 下午 10:56* @Description**/
public class MySubscriber implements Subscriber<String> {private Subscription subscription;@Overridepublic void onSubscribe(Subscription subscription) {this.subscription = subscription;subscription.request(5); // 一次请求5个数据}@Overridepublic void onNext(String s) {System.out.println("Received: " + s);// 每处理一个数据，继续请求一个数据subscription.request(1);}@Overridepublic void onError(Throwable t) {System.err.println("Error: " + t.getMessage());}@Overridepublic void onComplete() {System.out.println("All data processed");}
}

在这个例子中，Subscriber 控制每次只处理 5 个数据，然后根据处理速度继续请求。

4. Reactive-Streams 与 Reactor

Reactor 是 Spring 的响应式编程库，完全基于 Reactive-Streams 规范。它通过 Flux 和 Mono 两种 Publisher 来实现数据流的发布。

Mono：表示一个包含 0 或 1 个数据的异步流。
Flux：表示一个包含 0 到多个数据的异步流。

Reactor 的底层实现遵循了 Reactive-Streams 规范，并扩展了许多强大的操作符，用于流的转换、过滤、组合等操作。

例如，Reactor 中的一个简单数据流处理示例：

Flux.just("A", "B", "C").map(String::toLowerCase).subscribe(new MySubscriber());

生产者：Flux.just("A", "B", "C") 是生产者，它负责发布数据（即 "A", "B", "C"），形成一个包含这三个元素的异步数据流。Flux 是 Publisher 的实现。
消费者：subscribe(new MySubscriber()) 是消费者，它订阅了数据流并消费数据。System.out::println 作为 Subscriber，每接收到一个数据就执行打印操作。