Reactive Streams， flux， mono介绍之白话版

红太狼 2022-04-04 04:01 291阅读 0赞

# 目录 #

要解决的问题：

解决原理：

解决方法：

发布者Publisher

订阅者Subscriber

订阅对象Subscription

处理者Processor

源码简析

spring中响应式流的应用

服务端技术栈

响应式Http客户端

服务端修改

客户端webClient的使用方法

--------------------

响应式流(Reactive Streams)是以带非阻塞背压方式处理异步数据流的标准，提供一组最小化的接口，方法和协议来描述必要的操作和实体。

## **要解决的问题：** ##

系统之间高并发的大量数据流交互通常采用异步的发布-订阅模式。数据由发布者推送给订阅者的过程中，容易产生的一个问题是，当发布者即生产者产生的数据速度远远大于订阅者即消费者的消费速度时，消费者会承受巨大的资源压力(pressure)而有可能崩溃。

## **解决原理：** ##

为了解决以上问题，数据流的速度需要被控制，即流量控制(flow control)，以防止快速的数据流不会压垮目标。因此需要反压即背压(back pressure)，生产者和消费者之间需要通过实现一种背压机制来互操作。实现这种背压机制要求是异步非阻塞的，如果是同步阻塞的，消费者在处理数据时生产者必须等待，会产生性能问题。

## **解决方法：** ##

响应式流(Reactive Streams)通过定义一组实体，接口和互操作方法，给出了实现非阻塞背压的标准。第三方遵循这个标准来实现具体的解决方案，常见的有Reactor，RxJava，Akka Streams，Ratpack等。

该标准定义了四个接口：

*  ### **发布者Publisher** ###

interface Publisher<T> {
     void subscribe(Subscriber<? super T> subscriber);
    }

发布者只有一个方法，用来接受订阅者进行订阅(subscribe)。T代表发布者和订阅者之间传输的数据类型。它有两个实现类，Flux<T>和Mono<T>。通俗理解就是Flux可以触发零到多个事件。Mono最多只触发一个事件。

*  ### **订阅者Subscriber** ###

interface Subscriber<T> {
     void onSubscribe(Subscription s);
     void onNext(T t);
     void onError(Throwable t);
     void onComplete();
    }

订阅者有四个事件方法，分别在开启订阅、接收数据、发生错误和数据传输结束时被调用。

*  ### **订阅对象Subscription** ###

interface Subscription {
     void request(long n);
     void cancel();
    }

*  ### **处理者Processor** ###

interface Processor<T,R> extends Subscriber<T>, Publisher<R> {
    }

处理者既是发布者又是订阅者，用于在发布者和订阅者之间转换数据格式，把发布者的R类型数据转换为订阅者接受的T类型数据。处理者作为数据转换的中介不是必须的。

> 由以上的接口可以看出，核心在于订阅者可以通过request(long n)方法来控制接收的数据量，达到了实现背压的目的。

概念太抽象，举个栗子吧：

@Test
    public void testBackpressure() {
        Flux.just("one","two").log().subscribe(new SubscriberImpl());
    }
    
    class SubscriberImpl implements Subscriber<String> {
    
        @Override
        public void onSubscribe(Subscription s) {
            s.request(1);
        }
    
        @Override
        public void onNext(String s) {
            System.out.println(s);
        }
    
        @Override
        public void onError(Throwable t) {
        }
    
        @Override
        public void onComplete() {
            System.out.println("done");
        }
    }

来看上面的例子，订阅对象（Subscription  s）是发布者（Flux）和订阅者之间交互的操作对象，在发布者(Flux)通过subscribe方法加入订阅者时，会通过调用订阅者(Subscriber)的onSubscribe把订阅对象(Subscription)传给订阅者（Subscriber）。

订阅者（Subscriber）拿到订阅对象(Subscription)后，通过调用订阅对象的request方法，根据自身消费能力请求n条数据，或者调用cancel方法来停止接收数据（当调用cancel方法后则不会进行消费）。

订阅对象的request方法被调用时，会触发订阅者的onNext事件方法，把数据传输给订阅者。如果数据全部传输完成，则触发订阅者的onComplete事件方法。如果数据传输发生错误，则触发订阅者的onError事件方法。

log()方法顾名思义就是增加日志打印，上面的代码运行后输出如下：

18:47:00.808 [main] INFO reactor.Flux.Array.1 - | onSubscribe([Synchronous Fuseable] FluxArray.ArraySubscription)
    18:47:00.810 [main] INFO reactor.Flux.Array.1 - | request(1)
    18:47:00.811 [main] INFO reactor.Flux.Array.1 - | onNext(one)
    one

因为只请求了一条数据，并没有执行完，所以onComplete方法没有执行。

# 源码简析 #

Flux.just方法实际上是把需要发布的数据放在一个数组中（Flux类负责完成）

/**
    	 * Create a {@link Flux} that emits the provided elements and then completes.
    	 * <p>
    	 * <img class="marble" src="https://raw.githubusercontent.com/reactor/reactor-core/v3.1.0.RC1/src/docs/marble/justn.png" alt="">
    	 * <p>
    	 * @param data the elements to emit, as a vararg
    	 * @param <T> the emitted data type
    	 *
    	 * @return a new {@link Flux}
    	 */
    	@SafeVarargs
    	public static <T> Flux<T> just(T... data) {
    		return fromArray(data);
    	}

数组初始化完成后调用subscribe方法，参数为Subscriber接口的实现类

@SuppressWarnings("unchecked")
    	public static <T> void subscribe(CoreSubscriber<? super T> s, T[] array) {
    		if (array.length == 0) {
    			Operators.complete(s);
    			return;
    		}
    		if (s instanceof ConditionalSubscriber) {
    			s.onSubscribe(new ArrayConditionalSubscription<>((ConditionalSubscriber<? super T>) s, array));
    		}
    		else {
    			s.onSubscribe(new ArraySubscription<>(s, array));
    		}
    	}

最后跳转到FluxArray的subscribe方法中，调用CoreSubscriber接口的onSubscribe方法，因为SubscriberImpl实现了Subscriber接口，这里的s其实就是类SubscriberImpl的对象，于是代码跳转到实现类的onSubscribe方法中，如下：

class SubscriberImpl implements Subscriber<String> {
    
            public void onSubscribe(Subscription s) {
                s.request(1);
            }
    
            public void onNext(String s) {
                System.out.println(s);
            }
    
            public void onError(Throwable t) {}
    
            public void onComplete() {
                System.out.println("done");
            }
        }

执行request方法，跟踪代码，会跳转到FluxArray类的如下方法：

void slowPath(long n) {
    			final T[] a = array;
    			final int len = a.length;
    			final Subscriber<? super T> s = actual;
    
    			int i = index;
    			int e = 0;
    
    			for (; ; ) {
    				if (cancelled) {
    					return;
    				}
    
    				while (i != len && e != n) {
    					T t = a[i];
    
    					if (t == null) {
    						s.onError(new NullPointerException("The " + i + "th array element was null"));
    						return;
    					}
    
    					s.onNext(t);
    
    					if (cancelled) {
    						return;
    					}
    
    					i++;
    					e++;
    				}
    
    				if (i == len) {
    					s.onComplete();
    					return;
    				}
    
    				n = requested;
    
    				if (n == e) {
    					index = i;
    					n = REQUESTED.addAndGet(this, -e);
    					if (n == 0) {
    						return;
    					}
    					e = 0;
    				}
    			}
    		}

`我们注意到REQUESTED是个AtomicLongFieldUpdater对象，所以addAndGet是个原子操作，可以保证线程安全，同时，n=requested赋值语句中，requested是个long类型的变量，而且是volatile的，保证了在多线程当中的可见性。当然，这些操作都是为了保证在while循环中s.onNext操作的原子性。而且我们看到有个if语句，当i==len的时候才会执行onComplete方法，也就是当所有元素都传输完成才会执行onComplete方法，以上，就是响应式流的基本原理。`

需要使用响应式流需要引入下面两个依赖：

<dependency>
                <groupId>org.reactivestreams</groupId>
                <artifactId>reactive-streams</artifactId>
                <version>1.0.0</version>
            </dependency>
            <dependency>
                <groupId>io.projectreactor</groupId>
                <artifactId>reactor-core</artifactId>
                <version>3.1.0.RELEASE</version>
            </dependency>

reactive-streams中其实只有四个接口，也就是符合响应式流的四个接口，下面的reactor-core才是正经干活的实现类。

# spring中响应式流的应用 #

*  spring中的响应式流框架叫Spring WebFlux，是随Spring 5推出的响应式Web框架。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2phdmFlcl9sZWU_size_16_color_FFFFFF_t_70][]

## 1、服务端技术栈 ##

*  Spring提供了完整的支持响应式的服务端技术栈。
 *  如上图所示，左侧为基于spring-webmvc的技术栈，右侧为基于spring-webflux的技术栈，
 *  Spring WebFlux是基于响应式流的，因此可以用来建立异步的、非阻塞的、事件驱动的服务。它采用Reactor作为首选的响应式流的实现库，不过也提供了对RxJava的支持。
 *  由于响应式编程的特性，Spring WebFlux和Reactor底层需要支持异步的运行环境，比如Netty和Undertow；也可以运行在支持异步I/O的Servlet 3.1的容器之上，比如Tomcat（8.0.23及以上）和Jetty（9.0.4及以上）。
 *  从图的纵向上看，spring-webflux上层支持两种开发模式： 
 *  类似于Spring WebMVC的基于注解（@Controller、@RequestMapping）的开发模式；
 *  Java 8 lambda 风格的函数式开发模式。
 *  Spring WebFlux也支持响应式的Websocket服务端开发。

## 2、响应式Http客户端 ##

*  此外，Spring WebFlux也提供了一个响应式的Http客户端API WebClient。它可以用函数式的方式异步非阻塞地发起Http请求并处理响应。其底层也是由Netty提供的异步支持。
 *  我们可以把WebClient看做是响应式的RestTemplate，与后者相比，前者是非阻塞的，可以基于少量的线程处理更高的并发；
 *  可以使用Java 8 lambda表达式；
 *  支持异步的同时也可以支持同步的使用方式；
 *  可以通过数据流的方式与服务端进行双向通信。

### 服务端修改 ###

*  服务端改成webflux只需要将引用的jar包换成spring webFlux，并且将controller的返回值用发布者代替就可以将框架的技术栈换成响应式流，具体操作如下：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

替换为：

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>

@RestController
        public class HelloController {
    
            @GetMapping("/hello")
            public String hello() {
                return "Welcome to reactive world ~";
            }
        }

替换为：

@RestController
        public class HelloController {
    
            @GetMapping("/hello")
            public Mono<String> hello() { 
                return Mono.just("Welcome to reactive world ~"); 
            }
        }

Spring真是用心良苦，WebFlux提供了与之前WebMVC相同的一套注解来定义请求的处理，使得Spring使用者迁移到响应式开发方式的过程变得异常轻松。

虽然我们只修改了少量的代码，但是其实这个简单的项目已经脱胎换骨了。整个技术栈从命令式的、同步阻塞的【spring-webmvc + servlet + Tomcat】变成了响应式的、异步非阻塞的【spring-webflux + Reactor + Netty】。

Netty是一套异步的、事件驱动的网络应用程序框架和工具，能够开发高性能、高可靠性的网络服务器和客户端程序，因此与同样是异步的、事件驱动的响应式编程范式一拍即合。

> 在Java 7推出异步I/O库，以及Servlet3.1增加了对异步I/O的支持之后，Tomcat等Servlet容器也随后开始支持异步I/O，然后Spring WebMVC也增加了对Reactor库的支持，所以上边第4）步如果不是将spring-boot-starter-web替换为spring-boot-starter-WebFlux，而是增加reactor-core的依赖的话，仍然可以用注解的方式开发基于Tomcat的响应式应用。

### 客户端webClient的使用方法 ###

@Test
        public void webClientTest1() throws InterruptedException {
            WebClient webClient = WebClient.create("http://localhost:8080");    // 1
            Mono<String> resp = webClient
                    .get().uri("/hello")                                        // 2
                    .retrieve()                                                 // 3
                    .bodyToMono(String.class);                                  // 4
            resp.subscribe(System.out::println);                                // 5
            TimeUnit.SECONDS.sleep(1);                                          // 6
        }

1.  创建WebClient对象并指定baseUrl；
2.  HTTP GET；
3.  异步地获取response信息；
4.  将response body解析为字符串；
5.  打印出来；
6.  由于是异步的，我们将测试线程sleep 1秒确保拿到response

运行结果如下：

17:45:14.782 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClient - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] FLUSH
    17:45:14.784 [reactor-http-nio-4] DEBUG reactor.ipc.netty.channel.ChannelOperationsHandler - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] Writing object EmptyLastHttpContent
    17:45:14.784 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClient - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] WRITE: 0B
    17:45:14.784 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClient - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] FLUSH
    17:45:14.794 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClient - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] READ: 106B
             +-------------------------------------------------+
             |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
    +--------+-------------------------------------------------+----------------+
    |00000000| 48 54 54 50 2f 31 2e 31 20 32 30 30 20 4f 4b 0d |HTTP/1.1 200 OK.|
    |00000010| 0a 43 6f 6e 74 65 6e 74 2d 54 79 70 65 3a 20 74 |.Content-Type: t|
    |00000020| 65 78 74 2f 70 6c 61 69 6e 3b 63 68 61 72 73 65 |ext/plain;charse|
    |00000030| 74 3d 55 54 46 2d 38 0d 0a 43 6f 6e 74 65 6e 74 |t=UTF-8..Content|
    |00000040| 2d 4c 65 6e 67 74 68 3a 20 32 37 0d 0a 0d 0a 57 |-Length: 27....W|
    |00000050| 65 6c 63 6f 6d 65 20 74 6f 20 72 65 61 63 74 69 |elcome to reacti|
    |00000060| 76 65 20 77 6f 72 6c 64 20 7e                   |ve world ~      |
    +--------+-------------------------------------------------+----------------+
    17:45:14.795 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClientOperations - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] Received response (auto-read:false) : [Content-Type=text/plain;charset=UTF-8, Content-Length=27]
    17:45:14.796 [reactor-http-nio-4] DEBUG org.springframework.web.reactive.function.client.ExchangeFunctions - Response received, status: 200 OK
    17:45:14.808 [reactor-http-nio-4] DEBUG reactor.ipc.netty.channel.FluxReceive - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] Subscribing inbound receiver [pending: 0, cancelled:false, inboundDone: false]
    17:45:14.812 [reactor-http-nio-4] DEBUG reactor.ipc.netty.http.client.HttpClientOperations - [id: 0x5eb115be, L:/127.0.0.1:59899 - R:localhost/127.0.0.1:8080] Received last HTTP packet
    Welcome to reactive world ~

参考来源：

*  [http://www.pzyxw.cn/fanwen/2018/0906/121435.html][http_www.pzyxw.cn_fanwen_2018_0906_121435.html]
 *  [https://blog.csdn.net/get\_set/article/details/79480233][https_blog.csdn.net_get_set_article_details_79480233]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2phdmFlcl9sZWU_size_16_color_FFFFFF_t_70]: /images/20220404/f0744f4951504ba683f686a75b4affd8.png
[http_www.pzyxw.cn_fanwen_2018_0906_121435.html]: http://www.pzyxw.cn/fanwen/2018/0906/121435.html
[https_blog.csdn.net_get_set_article_details_79480233]: https://blog.csdn.net/get_set/article/details/79480233