当前位置：首页 > news >正文

spring响应式编程系列：总体流程

news 来源：原创 2025/8/15 4:45:42

示例

程序流程

just

new LambdaMonoSubscriber

new Operators.ScalarSubscription

onSubscribe

request

onNext

时序图

类图

数据发布者

MonoJust

数据订阅者

LambdaSubscriber

订阅的消息体

ScalarSubscription

想要了解响应式编程的总体流程，只要做到真正吃透一个简单的示例即可。

如下所示：

示例

首先，通过调用Mono.just创建一个单元素的数据发布者（Publisher）；

然后，通过调用mono.subscribe订阅数据发布者（Publisher）发布的数据。

如下所示：

// 创建一个包含数据的 Mono
Mono<String> mono = Mono.just("Hello, Reactive World!");
// 订阅并消费 Mono
mono.subscribe(System.out::println);

程序流程

点击Mono.just，如下所示：

just

public static <T> Mono<T> just(T data) {

return onAssembly(new MonoJust(data));

}

在这里，直接new一个MonoJust对象并返回。

点击示例里的mono.subscribe，如下所示：

public abstract class Mono<T> implements CorePublisher<T> {

... ...

public final Disposable subscribe(@Nullable Consumer<? super T> consumer, @Nullable Consumer<? super Throwable> errorConsumer, @Nullable Runnable completeConsumer, @Nullable Context initialContext) {

return (Disposable)this.subscribeWith(new LambdaMonoSubscriber(consumer, errorConsumer, completeConsumer, (Consumer)null, initialContext));

}

在这里，将示例里subscribe的参数作为LambdaMonoSubscriber的构造参数，然后new一个LambdaMonoSubscriber对象。

LambdaMonoSubscriber对象的初始化参数，如下所示：

new LambdaMonoSubscriber

final class LambdaMonoSubscriber<T> implements InnerConsumer<T>, Disposable {

final Consumer<? super T> consumer;

final Consumer<? super Throwable> errorConsumer;

final Runnable completeConsumer;

final Consumer<? super Subscription> subscriptionConsumer;

final Context initialContext;

volatile Subscription subscription;

... ...

LambdaMonoSubscriber(@Nullable Consumer<? super T> consumer, @Nullable Consumer<? super Throwable> errorConsumer, @Nullable Runnable completeConsumer, @Nullable Consumer<? super Subscription> subscriptionConsumer, @Nullable Context initialContext) {

this.consumer = consumer;

this.errorConsumer = errorConsumer;

this.completeConsumer = completeConsumer;

this.subscriptionConsumer = subscriptionConsumer;

this.initialContext = initialContext == null ? Context.empty() : initialContext;

}

final class MonoJust<T> extends Mono<T> implements ScalarCallable<T>, Fuseable, SourceProducer<T> {

... ...

public void subscribe(CoreSubscriber<? super T> actual) {

actual.onSubscribe(Operators.scalarSubscription(actual, this.value));

}

在这里，来到了MonoJust对象的subscribe方法，该方法调用了LambdaMonoSubscriber对象的onSubscribe方法；

同时，new一个Operators.ScalarSubscription对象，该对象封装了LambdaMonoSubscriber对象和数据发布者MonoJust发布的数据。

如下所示：

new Operators.ScalarSubscription

public static <T> Subscription scalarSubscription(CoreSubscriber<? super T> subscriber, T value, String stepName) {

return new Operators.ScalarSubscription(subscriber, value, stepName);

}

点击actual.onSubscribe，如下所示：

onSubscribe

final class LambdaMonoSubscriber<T> implements InnerConsumer<T>, Disposable {

... ...

public final void onSubscribe(Subscription s) {

if (Operators.validate(this.subscription, s)) {

this.subscription = s;

if (this.subscriptionConsumer != null) {

try {

this.subscriptionConsumer.accept(s);

} catch (Throwable var3) {

Exceptions.throwIfFatal(var3);

s.cancel();

this.onError(var3);

}

} else {

s.request(9223372036854775807L);

}

在这里，LambdaMonoSubscriber对象调用了Operators.ScalarSubscription对象的request方法。

如下所示：

request

static final class ScalarSubscription<T> implements SynchronousSubscription<T>, InnerProducer<T> {

public void request(long n) {

if (Operators.validate(n) && ONCE.compareAndSet(this, 0, 1)) {

Subscriber<? super T> a = this.actual;

a.onNext(this.value);

if (this.once != 2) {

a.onComplete();

}

在这里，Operators.ScalarSubscription对象又调用了LambdaMonoSubscriber对象的onNext方法。

LambdaMonoSubscriber对象的onNext方法如下所示：

onNext

final class LambdaMonoSubscriber<T> implements InnerConsumer<T>, Disposable {

public final void onNext(T x) {

Subscription s = (Subscription)S.getAndSet(this, Operators.cancelledSubscription());

if (s == Operators.cancelledSubscription()) {

Operators.onNextDropped(x, this.initialContext);

} else {

if (this.consumer != null) {

try {

this.consumer.accept(x);

} catch (Throwable var5) {

Exceptions.throwIfFatal(var5);

s.cancel();

this.doError(var5);

}

if (this.completeConsumer != null) {

try {

this.completeConsumer.run();

} catch (Throwable var4) {

Operators.onErrorDropped(var4, this.initialContext);

}

终于，在这里，调用了示例里subscribe()方法的回调函数了。

时序图

【说明】

Mono和MonoJust是数据发布者，LambdaMonoSubscriber是数据消费者，ScalarSubscription是订阅的消息；
类的设计还是比较清晰的，就是方法的调用显示有点绕。
数据发布者，提供了just方法来生成数据发布者（Publisher）；
数据订阅者，提供了onSubscribe和onNext方法来响应订阅事件和读取数据；
订阅的消息体，封装了数据订阅者和数据发布发布的数据，并且提供了request方法用来处理数据。
使用了观察者设计模式：LambdaMonoSubscriber是观察者模式中的观察者（Observer），它订阅（subscribe）一个发布者（MonoJust），MonoJust是观察者模式中的主题（Subject），它负责通知所有的 Subscriber。

类图

数据发布者

MonoJust

【说明】

Publisher

定义了接口：void subscribe(Subscriber<? super T> var1)。

CorePublisher

定义了接口：void subscribe(CoreSubscriber<? super T> subscriber)。

Mono

是一个抽象类，实现了数据发布者通用的各种功能。

比如：使用了工厂方法设计模式来创建诸如MonoJust、MonoCreate、MonoDefer、MonoError等各种具体的数据发布者。

MonoJust

是一个特定的数据发布者（Publisher），实现了接口void subscribe(CoreSubscriber<? super T> actual)。

数据订阅者

LambdaSubscriber

【说明】

Subscriber

定义了如下接口：onSubscribe、onNext、onError、onComplete。

CoreSubscriber

定义了如下接口：onSubscribe

LambdaMonoSubscriber

关联了consumer、errorConsumer、completeConsumer、subscriptionConsumer这些对象，以完成订阅相关的各种操作。

订阅的消息体

ScalarSubscription

【说明】

Subscription

提供了如下接口：void request(long var1)、void cancel();

ScalarSubscription

封装了数据订阅者和数据发布者发布的数据。

示例

程序流程

​​​​​​​just

subscribe

​​​​​​​new LambdaMonoSubscriber

​​​​​​​MonoJust.subscribe

​​​​​​​new Operators.ScalarSubscription

​​​​​​​onSubscribe

​​​​​​​request

​​​​​​​onNext

时序图

类图

数据发布者

MonoJust

数据订阅者

LambdaSubscriber

订阅的消息体

ScalarSubscription

相关文章：

just

new LambdaMonoSubscriber

MonoJust.subscribe

new Operators.ScalarSubscription

onSubscribe

request

onNext