Reactor响应式编程（Mono）
Posted
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Reactor响应式编程（Mono）相关的知识，希望对你有一定的参考价值。
参考技术A （先写目标：不要求自己会写，只要几个常见的写法自己能看明白不耽误查资料就行了，看这个是为了学过滤器里面的那些东西。）返回值，原来返回List<xxxx>的地方变成Flux<xxxx>，原来返回单个对象的地方变成Mono<xxxx>。
Flux就是List的一种变式，Mono表示单个对象。
（由于Mono和Flux这两个意思差不多，所以只针对于Mono就可以了）
首先是Mono.just()，直接由这个对象构造出一个Mono。
然后Mono.fromRunnable(），用一个线程来构建一个Mono。
flatMap（把一个Mono处理一下变成另一个Mono，意思是这后面可以接着加点什么什么，继续链式处理）
Map（这个Mono直接就变成了普通对象，后面接不上了）
then（这个只是时间上的先后，并不是逻辑上的处理先后，就是then处理前后并不是一个变量，then接收一个变量，然后then前面处理的那个就结束了，后面开始处理then接收的这个变量）
这个基本不用你管，你只要返回一个Mono对象，Spring自动就帮你处理后续的。
Java ProjectReactor 响应式编程 Mono 简单工作流程解析
前言
我们在查看 Spring Cloud 源码的时候，会发现已经引入了 Mono 或者 Flux 相关的代码，如果对这些代码不熟悉，就会觉得有些 Spring Cloud 源码将会变得晦涩难懂。Mono 和 Flux 为 ProjectReactor 响应式框架中的核心类。其相关概念可以参考 Flux、Mono、Reactor 实战（史上最全）和
响应式编程入门之 Project Reactor。我在参考了这些文章后，查看了相应的源码，这里是将自己的理解记录下来，希望可以帮助到初学者理解 ProjectReactor 。本文的目标是可以让大家理解以下者行代码的实现逻辑。
Mono.just("hello").map(e->e+" world").map(e->e+"!").subscribe(System.out::println);核心接口
本文使用的是 java 8 ,项目中需要引入以下依赖
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-core</artifactId>
<version>3.4.24</version>
</dependency>ProjectReactor 其核心思想观察者模式，定义了发布者和订阅者以及订阅三个接口。其核心接口如下
/**
* 当将一个 Subscriber 传递给 Publisher#subscribe(Subscriber) 方法的时候，Subscriber#onSubscribe(Subscription) 会被调用
*/
public interface Subscriber<T>
/**
* 在 Publisher#subscribe(Subscriber) 调用时，被调用.
* 他的职责是去调用 Subscription#request(long) 方法
*/
public void onSubscribe(Subscription s);
/**
* 当 Subscription#request(long)调用的时候被调用，数据T来自 Publisher
*/
public void onNext(T t);
/**
* 发生异常时被触发
*/
public void onError(Throwable t);
/**
* 正常完成时被触发
*/
public void onComplete();
/**
* Subscription 代表 Subscriber 订阅一个 Publisher 的生命周期，只能被一个 Subscriber 使用一次。
*/
public interface Subscription
/**
* 通常是 Subscriber#onSubscribe(Subscription)时被触发，然后调用 Subscriber#onNext(T)方法
*/
public void request(long n);
/**
* 停止发送数据和清理相关资源
*/
public void cancel();
/**
* 一个发布者可以发布无数个元素给订阅者，被多个 Subscriber 进行订阅。
*/
public interface Publisher<T>
/**
* 这个方法用于开始一个数据流，可以被调用多次，每次都会生成一个新的 Subscription 对象，每个 Subscription 对象只能被一个 Subscriber 使用。每个 Subscriber 最多订阅一次一个 Publisher。但是 Publisher 可以被多个 Subscriber 订阅。
*/
public void subscribe(Subscriber<? super T> s);
仔细看官方接口的文档，可以看到核心调用逻辑如下：
首先调用 Publisher#subscribe(Subscriber)方法，传入了一个 Subscriber。
然后 Subscriber#onSubscribe(Subscription)，传入了一个 Subscription。
然后 Subscription#request(long) 会被触发。
然后 Subscriber#onNext(T) 会被触发。
每个方法，都规定了具体的职责。
看到这里我就有一个疑问了:
为什么需要这样弯弯绕绕，而且需要引入一个中间的对象 Subscription 传递数据？
Publisher 提供数据，Subscriber 直接去获取不就好了吗？
这里先给出我的理解：
形成数据流，中间可以引入多个处理过程。
可以在 Subscriber 订阅的时候，整个处理流程才动起来。
这里先有一个印象，后续可以再回头看就理解了。
按照核心逻辑，我们对以上三个接口进行简单实现。
简单实现
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
public class Demo
static class MySubscription implements Subscription
private String data;
private Subscriber subscriber;
public MySubscription(String data, Subscriber subscriber)
this.data = data;
this.subscriber = subscriber;
private boolean isCanceled;
@Override
public void request(long l)
if (!isCanceled)
try
subscriber.onNext(data);
subscriber.onComplete();
catch (Exception e)
subscriber.onError(e);
@Override
public void cancel()
isCanceled = true;
static class MyPublisher implements Publisher
private String data;
@Override
public void subscribe(Subscriber subscriber)
subscriber.onSubscribe(new MySubscription(data, subscriber));
public static Publisher just(String a)
MyPublisher myPublisher = new MyPublisher();
myPublisher.data = a;
return myPublisher;
static class MySubscriber implements Subscriber
@Override
public void onSubscribe(Subscription subscription)
subscription.request(1L);
@Override
public void onNext(Object o)
System.out.println(o);
@Override
public void onError(Throwable throwable)
System.out.println("error");
@Override
public void onComplete()
System.out.println("completed");
public static void main(String[] args)
MyPublisher.just("MyPublisher1").subscribe(new MySubscriber());
// 打印如下：
// MyPublisher1
//completed
是不是特别简单，一个简单的发布订阅流程就完成了。
但很明显扩展性不强，比如我想在 MyPublisher 和 MySubscriber 之间，做一些不限次数的处理逻辑，而且写法跟Stream类似，怎么办呢？
为了完成这个逻辑，需要引入一些操作类（operator）的对象，包括 Publisher，Subscriber 以及 Subscription 的实现类。
引入中间操作类的实现
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import java.util.function.Function;
public class OperatorPublisherTest
static class MySubscription implements Subscription
private String data;
private Subscriber subscriber;
public MySubscription(String data, Subscriber subscriber)
this.data = data;
this.subscriber = subscriber;
private boolean isCanceled;
@Override
public void request(long l)
System.out.println("MySubscription: 调用 MySubscription#request(long)");
if (!isCanceled)
try
subscriber.onNext(data);
subscriber.onComplete();
catch (Exception e)
subscriber.onError(e);
@Override
public void cancel()
isCanceled = true;
/**
* 抽象类，用于引入操作类的 Publisher
*/
static abstract class AbstractPublisher implements Publisher
public OperatorPublisher operator(Function function,String name )
return new OperatorPublisher(this,function,name);
/**
* SourcePublisher 用于获取最初始的数据
*/
static class SourcePublisher extends AbstractPublisher
private String name;
private String data;
public SourcePublisher(String name)
this.name = name;
println("生成 SourcePublisher ");
@Override
public void subscribe(Subscriber subscriber)
println("SourcePublisher#subscribe(Subscriber), 生成 MySubscription 对象");
subscriber.onSubscribe(new MySubscription(data, subscriber));
public static AbstractPublisher just(String a)
SourcePublisher myPublisher = new SourcePublisher("数据源Publisher");
myPublisher.data = a;
return myPublisher;
void println(String context)
System.out.println(name+": "+context);
static class OperatorPublisher extends AbstractPublisher
private String name;
private Publisher source;
private OperatorPublisher prePublisher;
Function mapper;
public OperatorPublisher(Publisher prePublisher,Function function,String name)
this.name = name;
this.source = prePublisher;
if (prePublisher instanceof OperatorPublisher)
this.prePublisher = (OperatorPublisher) prePublisher;
mapper = function;
print("生成 OperatorPublisher 对象");
@Override
public void subscribe(Subscriber s)
print("订阅：OperatorPublisher#subscribe(Subscriber)");
OperatorPublisher currentPublisher = this;
Subscriber currentSubscriber = s;
while (true)
currentSubscriber = currentPublisher.subscribeOrReturn(currentSubscriber);
OperatorPublisher nextPublisher = currentPublisher.prePublisher;
if (nextPublisher == null)
currentPublisher.source.subscribe(currentSubscriber);
return;
currentPublisher = nextPublisher;
public OperatorSubscriptionSubscriber subscribeOrReturn(Subscriber s)
print("OperatorPublisher#subscribeOrReturn(Subscriber), 生成 OperatorSubscriptionSubscriber 对象");
return new OperatorSubscriptionSubscriber(mapper,s,name);
public void print(String text)
System.out.println(name+": "+text);
static class OperatorSubscriptionSubscriber implements Subscriber,Subscription
private String name;
private Subscriber preSubscriber;
private Function function;
private Subscription preSubscription;
public OperatorSubscriptionSubscriber(Function f, Subscriber preSubscriber, String name)
this.function = f;
this.preSubscriber = preSubscriber;
this.name = name+"_OperatorSubscriptionSubscriber";
private boolean isCanceled;
@Override
public void request(long l)
println("OperatorSubscriptionSubscriber#request(long)");
preSubscription.request(l);
@Override
public void onSubscribe(Subscription preSubscription)
println("OperatorSubscriptionSubscriber#onSubscribe(Subscription)");
this.preSubscription = preSubscription;
preSubscriber.onSubscribe(this);
@Override
public void onNext(Object o)
println("OperatorSubscriptionSubscriber#onNext(Object)");
Object apply = function.apply(o);
println("处理后的值为："+apply);
preSubscriber.onNext(apply);
@Override
public void onError(Throwable t)
preSubscriber.onError(t);
@Override
public void onComplete()
preSubscriber.onComplete();
@Override
public void cancel() isCanceled = true;
void println(String context)
System.out.println(name+": "+context);
static class MySubscriber implements Subscriber
private String name = "最终订阅者MySubscriber：";
@Override
public void onSubscribe(Subscription subscription)
println("MySubscriber#onSubscribe(Subscription)");
subscription.request(1L);
@Override
public void onNext(Object o)
println("最终订阅者MySubscriber: MySubscriber onNext");
println("最终订阅者MySubscriber：结果输出 "+o);
@Override
public void onError(Throwable throwable)
println("error");
@Override
public void onComplete()
println("MySubscriber#onComplete()");
void println(String context)
System.out.println(name+context);
public static void main(String[] args)
SourcePublisher.just("MyPublisher1")// 生成 SourcePublisher
.operator(e->e+" operator1 ","操作对象1")// 生成 OperatorPublisher
.operator(e->e+" operator2 ","操作对象2") // 生成 OperatorPublisher
.subscribe(new MySubscriber()); // 生成了多个嵌套的 Subscriber 回到 SourcePublisher 进行 onSubscribe 调用链
涉及的接口总结如下
接口 | 普通实现 | 操作类实现 |
Publisher | SourcePublisher | OperatorPublisher |
Subscriber | MySubscriber | OperatorSubscriptionSubscriber |
Subscription | MySubscription | OperatorSubscriptionSubscriber |
接口对比如下
逻辑如下：
引入了操作者的概念(OperatorPublisher)，本身是一个 Publisher ，可以通过 source 和 parentPublisher 变量将所有的 Publisher 组织成为一个 Publisher 单向链表，注意的是OperatorPublisher还包含一个函数式接口Function，这个是真正处理数据的地方。
引入了新的操作类型的 Subscriber（同时也是一个 Subscription），当 Publisher 链表被订阅的时候启动遍历 Publisher ，在遍历的过程中也形成了一个 Subscriber 的链表，链表最后一个 Subscriber 为真正的订阅者。
当遍历至 Publisher 链表最后一个元素的时候， Subscriber 的队列也已经构建完成。
调用最后一个 Publisher 的 subscribe 方法，执行 onSubscribe 方法，这个方法又会将 Subscriber 的链表遍历一遍，形成一个 Subscription 链表。
当 Subscriber 的链表遍历至最后一个的时候，通过调用 Subscriber 的 onSubscribe 方法，将会执行 subscription#request(long)，这里也是一个遍历的方法，遍历 Subscription 到最后一个元素。
执行最后一个 Subscription 的 request 方法，依次执行 Subscriber 链表中的 Subscriber 的 onNext方法。
在操作类型的 Subscriber 中，含有对数据的处理逻辑，数据依次被处理后，最终的 Subscriber，完成最后的输出。
流程图如下：
打印效果如下（为了方便描述加上了行号）：
数据源 Publisher: 生成 SourcePublisher
操作对象1: 生成 OperatorPublisher 对象
操作对象2: 生成 OperatorPublisher 对象
操作对象2: 订阅：OperatorPublisher#subscribe(Subscriber)
操作对象2: OperatorPublisher#subscribeOrReturn(Subscriber), 生成 OperatorSubscriptionSubscriber 对象
操作对象1: OperatorPublisher#subscribeOrReturn(Subscriber), 生成 OperatorSubscriptionSubscriber 对象
数据源Publisher: SourcePublisher#subscribe(Subscriber), 生成 MySubscription 对象
操作对象1_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber#onSubscribe(Subscription)
操作对象2_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber#onSubscribe(Subscription)
最终订阅者MySubscriber：MySubscriber#onSubscribe(Subscription)
操作对象2_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber#request(long)
操作对象1_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber#request(long)
MySubscription: 调用 MySubscription#request(long)
操作对象1_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber#onNext(Object)
操作对象1_OperatorSubscriptionSubscriber: 处理后的值为：MyPublisher1 operator1
操作对象2_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber#onNext(Object)
操作对象2_OperatorSubscriptionSubscriber: 处理后的值为：MyPublisher1 operator1 operator2
最终订阅者MySubscriber：最终订阅者MySubscriber: MySubscriber onNext
最终订阅者MySubscriber：最终订阅者MySubscriber：结果输出 MyPublisher1 operator1 operator2
最终订阅者MySubscriber：MySubscriber#onComplete()
以上，一个完整的发布订阅逻辑已经处理完成，通过查看流程，可以看到
Publisher 在调用 subscribe 方法之前，只是构建了一个 Publisher 链表，没有发生任何数据处理逻辑。
查看 Publisher 是不存在任何状态的，也就是说，这个 Publisher 链表是可以重复使用的。
当调用了 subscribe 方法之后，数据处理逻辑开始开动，而且查看 Subscriber，
Mono 的工作流程
当你看完前面的逻辑并已经理解了之后，查看 Mono 的工作流程就会变得比较简单
首先一行最简单的代码如下：
import reactor.core.publisher.Mono;
public class MonoTest
public static void main(String[] args)
Mono.just("hello").map(e->e+" world").map(e->e+"!").subscribe(System.out::println);
首先，查看Mono类，发现他是一个抽象类
public abstract class Mono implements CorePublisher
是一个 Publisher 的实现类，当他调用 just(T) 方法时，会返回一个 MonoJust，MonoJust 继承了 Mono。
public static <T> Mono<T> just(T data)
return onAssembly(new MonoJust<>(data));
然后调用 map(Function) 时，会返回一个 MonoMapFuseable ，MonoMapFuseable 也是继承了 Mono。继续调用 map(Function)，也是如此。
public final <R> Mono<R> map(Function<? super T, ? extends R> mapper)
if (this instanceof Fuseable)
return onAssembly(new MonoMapFuseable<>(this, mapper));
return onAssembly(new MonoMap<>(this, mapper));
至此，Publisher 的两个实现类已经出现 MonoJust 和 MonoMapFuseable。MonoJust 负责存储原始数据，MonoMapFuseable 用于处理数据，并负责构建 Publisher 链表。看下两个类的实现
final class MonoJust<T>
extends Mono<T>
implements Fuseable.ScalarCallable<T>, Fuseable, SourceProducer<T>
final T value;
// 将数据存储在 value 变量中。
MonoJust(T value)
this.value = Objects.requireNonNull(value, "value");
@Override
public void subscribe(CoreSubscriber<? super T> actual)
actual.onSubscribe(Operators.scalarSubscription(actual, value));
final class MonoMapFuseable<T, R> extends InternalMonoOperator<T, R>
implements Fuseable
final Function<? super T, ? extends R> mapper;
MonoMapFuseable(Mono<? extends T> source, Function<? super T, ? extends R> mapper)
super(source);
this.mapper = Objects.requireNonNull(mapper, "mapper");
// 为了方便理解，将 subscribe(CoreSubscriber)方法从 父类 InternalMonoOperator 搬到了这里。
@Override
@SuppressWarnings("unchecked")
public final void subscribe(CoreSubscriber<? super O> subscriber)
OptimizableOperator operator = this;
try
while (true)
subscriber = operator.subscribeOrReturn(subscriber);
if (subscriber == null)
// null means "I will subscribe myself", returning...
return;
OptimizableOperator newSource = operator.nextOptimizableSource();
if (newSource == null)
operator.source().subscribe(subscriber);
return;
operator = newSource;
catch (Throwable e)
Operators.reportThrowInSubscribe(subscriber, e);
return;
@Override
@SuppressWarnings("unchecked")
public CoreSubscriber<? super T> subscribeOrReturn(CoreSubscriber<? super R> actual)
if (actual instanceof ConditionalSubscriber)
ConditionalSubscriber<? super R> cs = (ConditionalSubscriber<? super R>) actual;
return new FluxMapFuseable.MapFuseableConditionalSubscriber<>(cs, mapper);
return new FluxMapFuseable.MapFuseableSubscriber<>(actual, mapper);
注意在 subscribeOrReturn 方法中返回了一个 MapFuseableConditionalSubscriber
可以看到 MapFuseableConditionalSubscriber 既是一个 Subscrition 也是一个 Subscriber。用来构建 Subscriber 链表和 Subscrition 链表，而它存有 MonoMapFuseable 传进来的 Function<? super T, ? extends R> mapper ，用于数据处理。其核心逻辑如下，
static final class MapFuseableConditionalSubscriber<T, R>
implements ConditionalSubscriber<T>, InnerOperator<T, R>,
QueueSubscription<R>
final ConditionalSubscriber<? super R> actual;
final Function<? super T, ? extends R> mapper;
boolean done;
QueueSubscription<T> s;
MapFuseableConditionalSubscriber(ConditionalSubscriber<? super R> actual,
Function<? super T, ? extends R> mapper)
this.actual = actual;
this.mapper = mapper;
@SuppressWarnings("unchecked")
@Override
public void onSubscribe(Subscription s)
if (Operators.validate(this.s, s))
this.s = (QueueSubscription<T>) s;
actual.onSubscribe(this);
@Override
public void onNext(T t)
if (sourceMode == ASYNC)
actual.onNext(null);
else
R v;
try
v = mapper.apply(t);
if (v == null)
throw new NullPointerException("The mapper [" + mapper.getClass().getName() + "] returned a null value.");
catch (Throwable e)
return;
actual.onNext(v);
@Override
public void request(long n)
s.request(n);
当调用 .subscribe(System.out::println) 的时候，这里传入的是一个方法引用。调用的是父类Mono的多个重载方法
public final Disposable subscribe(Consumer<? super T> consumer)
Objects.requireNonNull(consumer, "consumer");
return subscribe(consumer, null, null);
public final Disposable subscribe(
@Nullable Consumer<? super T> consumer,
@Nullable Consumer<? super Throwable> errorConsumer,
@Nullable Runnable completeConsumer)
return subscribe(consumer, errorConsumer, completeConsumer, (Context) null);
public final Disposable subscribe(
@Nullable Consumer<? super T> consumer,
@Nullable Consumer<? super Throwable> errorConsumer,
@Nullable Runnable completeConsumer,
@Nullable Context initialContext)
return subscribeWith(new LambdaMonoSubscriber<>(consumer, errorConsumer,
completeConsumer, null, initialContext));
public final <E extends Subscriber<? super T>> E subscribeWith(E subscriber)
// 这里将会调用子类的重写 subscribe 方法。
subscribe(subscriber);
return subscriber;
传入的 Consumer 被封装成为了 LambdaMonoSubscriber，最终在 onNext 方法中调用 Consumer 的逻辑。
final class LambdaMonoSubscriber<T> implements InnerConsumer<T>, Disposable
// 两个核心方法
@Override
public final void onSubscribe(Subscription s)
if (Operators.validate(subscription, s))
this.subscription = s;
if (subscriptionConsumer != null)
//。。。
else
s.request(Long.MAX_VALUE);
@Override
public final void onNext(T x)
if (consumer != null)
try
consumer.accept(x);
catch (Throwable t)
Exceptions.throwIfFatal(t);
s.cancel();
doError(t);
// 省略其他。。。
以上就是所有的关键类，跟我们前面的实现逻辑是一致的，流程也是一致，因此不再做过多的描述。
总结
本文通过自己实现接口的方式，揭示了响应式编程核心类 Mono 的基本原理。其核心接口为三个 Publisher,Subscriber,Subsription。核心方法为 Publisher#onSubscribe(Subscription),Subscriber#onSubscribe(Subscription),Subsription#request(long), Subscriber#onNext(T)，围绕Publisher的构造函数以及这四个方法，依次构建了Publisher 链表， Subscriber 链表和 Subsription 链表。通过这三个链表构建和遍历，完成了数据的发布，订阅和流式处理逻辑。
但是本文未涉及如何完成"响应式"，这个在后续的文章中会涉及。
参考
以上是关于Reactor响应式编程（Mono）的主要内容，如果未能解决你的问题，请参考以下文章
学习响应式编程 Reactor - reactor 转换类操作符