Java ProjectReactor 响应式编程 Mono 简单工作流程解析

青旅半醒 2024-03-27 11:29 24阅读 0赞

## 前言 ##

我们在查看 Spring Cloud 源码的时候，会发现已经引入了 Mono 或者 Flux 相关的代码，如果对这些代码不熟悉，就会觉得有些 Spring Cloud 源码将会变得晦涩难懂。Mono 和 Flux 为 ProjectReactor 响应式框架中的核心类。其相关概念可以参考 [Flux、Mono、Reactor 实战（史上最全）][Flux_Mono_Reactor]和

[响应式编程入门之 Project 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);
 }

仔细看官方接口的文档，可以看到核心调用逻辑如下：

1.  首先调用 Publisher\#subscribe(Subscriber)方法，传入了一个 Subscriber。

1.  然后 Subscriber\#onSubscribe(Subscription)，传入了一个 Subscription。

1.  然后 Subscription\#request(long) 会被触发。

1.  然后 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 调用链
        }
    }

涉及的接口总结如下

<table style="table-layout:fixed;width:28px;"> 
 <tbody> 
 <tr> 
 <td style="width:55px;vertical-align:top;height:52px;">接口</td> 
 <td style="width:85px;vertical-align:top;height:52px;">普通实现</td> 
 <td style="width:100px;vertical-align:top;height:52px;">操作类实现</td> 
 </tr> 
 <tr> 
 <td style="width:95px;vertical-align:top;height:52px;">Publisher</td> 
 <td style="width:155px;vertical-align:top;height:52px;">SourcePublisher</td> 
 <td style="width:170px;vertical-align:top;height:52px;">OperatorPublisher</td> 
 </tr> 
 <tr> 
 <td style="width:110px;vertical-align:top;height:52px;">Subscriber</td> 
 <td style="width:125px;vertical-align:top;height:52px;">MySubscriber</td> 
 <td style="width:275px;vertical-align:top;height:52px;">OperatorSubscriptionSubscriber</td> 
 </tr> 
 <tr> 
 <td style="width:125px;vertical-align:top;height:52px;">Subscription</td> 
 <td style="width:140px;vertical-align:top;height:52px;">MySubscription</td> 
 <td style="width:275px;vertical-align:top;height:52px;">OperatorSubscriptionSubscriber</td> 
 </tr> 
 </tbody> 
</table>

接口对比如下

![bc1d387951e722b2103dd7fbe90bf469.png][]

逻辑如下：

> 引入了操作者的概念(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，完成最后的输出。

流程图如下：

![0074fd9df099aa23db34e5c9c25dcf52.png][]

打印效果如下（为了方便描述加上了行号）：

1.  数据源 Publisher: 生成 SourcePublisher

1.  操作对象1: 生成 OperatorPublisher 对象

1.  操作对象2: 生成 OperatorPublisher 对象

1.  操作对象2: 订阅：OperatorPublisher\#subscribe(Subscriber)

1.  操作对象2: OperatorPublisher\#subscribeOrReturn(Subscriber), 生成 OperatorSubscriptionSubscriber 对象

1.  操作对象1: OperatorPublisher\#subscribeOrReturn(Subscriber), 生成 OperatorSubscriptionSubscriber 对象

1.  数据源Publisher: SourcePublisher\#subscribe(Subscriber), 生成 MySubscription 对象

1.  操作对象1\_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber\#onSubscribe(Subscription)

1.  操作对象2\_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber\#onSubscribe(Subscription)

1.  最终订阅者MySubscriber：MySubscriber\#onSubscribe(Subscription)

1.  操作对象2\_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber\#request(long)

1.  操作对象1\_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber\#request(long)

1.  MySubscription: 调用 MySubscription\#request(long)

1.  操作对象1\_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber\#onNext(Object)

1.  操作对象1\_OperatorSubscriptionSubscriber: 处理后的值为：MyPublisher1 operator1

1.  操作对象2\_OperatorSubscriptionSubscriber: OperatorSubscriptionSubscriber\#onNext(Object)

1.  操作对象2\_OperatorSubscriptionSubscriber: 处理后的值为：MyPublisher1 operator1 operator2

1.  最终订阅者MySubscriber：最终订阅者MySubscriber: MySubscriber onNext

1.  最终订阅者MySubscriber：最终订阅者MySubscriber：结果输出 MyPublisher1 operator1 operator2

1.  最终订阅者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 \{\}  
>

![fad54ed81ec8cf10a0544f75d12e7a7e.png][]

是一个 Publisher 的实现类，当他调用 just(T) 方法时，会返回一个 MonoJust，MonoJust 继承了 Mono。

public static <T> Mono<T> just(T data) {
 		return onAssembly(new MonoJust<>(data));
 	}

![a704d02e84ba26c89af0613e6dd67b59.png][]

然后调用 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));
 	}

![f89e6c287486588279457855840ef07b.png][]

至此，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

![af560dfbd05762f4440112341eeb4b06.png][]

可以看到 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 链表。通过这三个链表构建和遍历，完成了数据的发布，订阅和流式处理逻辑。

但是本文未涉及如何完成"响应式"，这个在后续的文章中会涉及。

## 参考 ##

[projectreactor官方网站][projectreactor]

[官方文档][Link 1]

[Flux、Mono、Reactor 实战（史上最全）][Flux_Mono_Reactor]

[二、subscribeOn和publishOn源码解析][subscribeOn_publishOn]

[响应式编程入门之 Project Reactor][Project Reactor]

[Flux_Mono_Reactor]: https://blog.csdn.net/crazymakercircle/article/details/124120506?spm=1001.2101.3001.6650.3&utm_medium=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-3-124120506-blog-124292452.pc_relevant_3mothn_strategy_recovery&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2%7Edefault%7ECTRLIST%7ERate-3-124120506-blog-124292452.pc_relevant_3mothn_strategy_recovery&utm_relevant_index=6
[Project Reactor]: https://www.jianshu.com/p/d7fa576e50da
[bc1d387951e722b2103dd7fbe90bf469.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/5c734e2f4d624da499eef11090a3a6ac.png
[0074fd9df099aa23db34e5c9c25dcf52.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/4de0cc9ccfb14039b22eb8aa6b545373.png
[fad54ed81ec8cf10a0544f75d12e7a7e.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/2c5e1d7e140143f9a5f3ddbfdb1cc529.png
[a704d02e84ba26c89af0613e6dd67b59.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/4f7eef5d860946398939147cb4c6b799.png
[f89e6c287486588279457855840ef07b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/bdf6bbe8711b47f2b47fbc17c844e179.png
[af560dfbd05762f4440112341eeb4b06.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/27/2ad555b3838f451db8604107589fc655.png
[projectreactor]: https://projectreactor.io/
[Link 1]: https://projectreactor.io/docs/core/release/reference/index.html
[subscribeOn_publishOn]: https://blog.csdn.net/chuxue1989/article/details/105128129/