首页 > 基础资料博客日记

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

2023-07-24 13:42:02基础资料围观226次

本篇文章分享Java ProjectReactor 响应式编程 Mono 简单工作流程解析，对你有帮助的话记得收藏一下，看Java资料网收获更多编程知识

前言

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

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

参考

projectreactor官方网站

官方文档

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

二、subscribeOn和publishOn源码解析

响应式编程入门之 Project Reactor

文章来源:https://blog.csdn.net/huachangzai/article/details/128844427
本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若内容造成侵权/违法违规/事实不符，请联系邮箱：jacktools123@163.com进行投诉反馈，一经查实，立即删除！

标签：

上一篇：Lombok报错：class lombok.javac.apt.LombokProcessor
下一篇：Java Base64 和 File 之间互转

首页 > 基础资料博客日记

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

前言

核心接口

简单实现

引入中间操作类的实现

Mono 的工作流程

总结

参考

相关文章

最新发布

点击排行

本站推荐

标签云

首页 > 基础资料 博客日记

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

前言

核心接口

简单实现

引入中间操作类的实现

Mono 的工作流程

总结

参考

相关文章

最新发布

点击排行

本站推荐

标签云

首页 > 基础资料博客日记