subscribe是什么意思

Spring5中Reactor编程的Mono与Flux类源码解读

Spring5的函数式编程使⽤了ProjectReactor⼯程的类，使⽤最为多的就是Mono和Flux类型，其中Mono是针对0到1个元素进⾏操

作，Flux是针对多个元素进⾏操作。要使⽤这两个类前提是⾃⼰要有jdk8，Lambda，函数式编程的基础，否则请先学习了以上知识再来接

触Spring5的Mono和FLux。

我们针对Mono类的⼀⾏代码，来进⾏源码分析与讲解。代码如下：

("hello").subscribe(::println);

这⾏代码是⽣成有⼀个元素的Mono类，并打印“hello”字符串。

⾸先，通过()⽅法⽣成了Mono类

publicstaticMonojust(Tdata){

returnonAsmbly(newMonoJust<>(data));

}

以上代码可以看到是创建了⼀个MonoJust类型，这个类和Mono，FLux类型⼀样在reactor-core包中，其实Mono的每个⽅法都会对应⽣

成⼀个Mono的⼦类，其⼦类很多，如下：

截图只是列举了⼀部分，Mono采⽤这种每个⽅法都⽣成⼀个类的⽅式是与jdk8中的Stream流⽔线的最⼤区别，⽬的是为了重⽤任意阶段

的结果，且其所有⼦类都实现了Mono类的⽅法：

publicabstractvoidsubscribe(CoreSubscriber<?superT>actual);

这个⽅法我们后⾯会讲，它的作⽤就是处理后⾯的函数式的逻辑。

在返回MonoJust类型后，我们调⽤了subscribe(::println)⽅法，这个⽅法在Mono类

有具体实现，⽅法如下：

publicfinalDisposablesubscribe(Consumer<?superT>consumer){

eNonNull(consumer,"consumer");

returnsubscribe(consumer,null,null);

}

这个⽅法是所有Mono的⼦类执⾏subscribe(CoreSubscriber<?superT>actual)⽅法的⼊⼝，我们进⼊看下，执⾏到了如下⽅法：

publicfinalDisposablesubscribe(

@NullableConsumer<?superT>consumer,

@NullableConsumer<?superThrowable>errorConsumer,

@NullableRunnablecompleteConsumer,

@NullableConsumer<?superSubscription>subscriptionConsumer){

returnsubscribeWith(newLambdaMonoSubscriber<>(consumer,errorConsumer,

completeConsumer,subscriptionConsumer));

}

我们看到有四个参数，根据字⾯意思即可理解，lambda表达式是第⼀个参数consumer，可以看到执⾏Lambda的⽅法时候创建了⼀个类

型LambdaMonoSubscriber，这个类型封装了封装了四个参数。

后⾯进⼊此⽅法：

@Override

publicfinalvoidsubscribe(Subscriber<?superT>actual){

onLastAsmbly(this).subscribe(Subscriber(actual));

}

这个⽅法是重写了⽗类Publisher的suscribe⽅法，我们看到代码做了类型转换，通过类型转换就转换为了可以调⽤MonoJust中的

subscribe⽅法了，MonoJust总的subscribe⽅法如下，所有的Mono⼦类都重写了此⽅法：

@Override

publicvoidsubscribe(CoreSubscriber<?superT>actual){

cribe(Subscription(actual,value));

}

然后通过LambdaMonoSubscriber的onSubscribe⽅法在调⽤Operators的request⽅法，再调⽤

此LambdaMonoSubscriber的onNext⽅法，最后调⽤了我们⾃⼰写的Lambda表达式，如下：

@Override

publicfinalvoidonNext(Tx){

Subscriptions=Set(this,ledSubscription());

if(s==ledSubscription()){

Dropped(x,());

return;

}

if(consumer!=null){

try{

(x);

}

consumer就是我们传⼊进来的Lambda表达式，这个⽅法继承⾃Suscriber类

通过这种⽅式，Publisher发送的subscribe就被Subscriber类消费掉了。其实全程都是⼀个单线程的操作。借鉴了消费订阅模式。

关于Mono的⽅法很多，在此只是举了⼀个简单的例⼦，其他的⽅法也可以通过类似的⽅式去研究。

在此处有⼀篇