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);

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

  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 调用链
    

涉及的接口总结如下

接口

普通实现

操作类实现

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,完成最后的输出。

流程图如下:

打印效果如下(为了方便描述加上了行号):

  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

是一个 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

以上是关于Reactor响应式编程(Mono)的主要内容,如果未能解决你的问题,请参考以下文章

响应式编程详解,带你熟悉Reactor响应式编程

我对响应式编程中Mono和Flux的理解

学习响应式编程 Reactor - reactor 转换类操作符

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

学习推荐5:Spring响应式编程实战

响应式编程系列:什么是响应式编程?reactor入门