响应式编程-基本介绍

什么是响应式编程

响应式宣言（The Reactive Manifesto）对响应式系统进行了定义：响应式系统是具备以下特质即时响应性（Responsive）、回弹性（Resilient）、弹性（Elastic）以及消息驱动（Message Driven）的系统。

响应式是一种思维模式，核心在于异步和非阻塞。最初起源于IO模型的Reactor模型，并形成理念产生了各类响应式框架，贯穿整个系统使用的编程模式，从系统接受输入开始到完成完整的系统功能输出结果的完整过程中都是响应式的。

图1 主从多线程模型的Reactor线程模型

响应式的底层逻辑如下：阻塞是非常浪费和影响性能的，要构建一个无阻塞的系统。响应式编程有三个最重要的特点：1. 事件驱动的处理模式。 2. 只有事件/消息真正被订阅时才会执行，在此之前都是对事件/消息处理流的一种编排。3. 事件/消息处理流可以对背压进行管理。

理解响应式编程：

以监听者模式来说，当需要发布一个事件时，通常会调用所欲listener事件处理接口

for(Listener listener : listeners) {

listener.onEvent(event);

}

以上例子中如果某个listener功能发生阻塞，将造成发布者线程阻塞。为改善阻塞，引入多线程/异步的方式：

ExecutorService executor = Executors.newCachedThreadPool();

for(Listener listener : listeners) {

executor.execute(()->{

listener.onEvent(event);

});

}

在上面使用线程池的情况下，线程数量受服务器限制，当事件大量增多是，可能会发生线程池处理队列满而发生阻塞的情况，如网关中涌入大量请求时的场景，因此即使每一个连接或每一个请求都做异步处理，调度线程总会在某时刻无法获取空闲线程而阻塞事件分发线程。
Listener处理事件如果是响应式编程， listener将对事件的处理编排成一个执行流并返回，但不会执行他，只有当执行流被订阅时才真正执行。

比如 onEvent将返回一个Mono或Flux对象。

Mono<String> result = Mono.fromSupplier(this::test);

响应式和多线程、异步的差别在哪：
1. 响应式是惰性的，也就是在被订阅之前不会执行任何操作。而Feature或线程池总是会执行某些操作，比如放入任务队列，申请线程等。
2. 当收到响应式的发布者对象时，由调用者可以将其封装在响应式链条中，从而构建一个完整的响应链。
3. 当需要订阅时（真正执行时），可以方便的使用Scheduler进行异步订阅处理。

背压是指来自后端的压力，当消息消费者的消费能力无法跟上消息发送者的能力，造成消息的积压，甚至可能由于持续的压力造成消息消费者的崩溃。类似于DDOS拒绝服务攻击的概念。异步消息驱动的系统需要能够进行背压管理，如引入消息中间件，消息将安全存储在消息中间件中，直到被可靠消费。而在响应式编程中也需要提供背压的处理机制避免消息积压和系统阻塞。

响应式编程的使用场景

响应式编程并不会提高程序的运行速度，由于存在一定的学习门和思维模式转变的原因，在业务系统的开发中并没有显示器重要性而得到应用。但在IO密集型的适用场景应用中得到了广泛的应用，如网关应用：zuul2， spring cloud gateway等。

Java 响应式流规范

Reactive-Streams(https://www.reactive-streams.org/) 是Java语言的响应式流编程模型，并已被JDK9采用，以java.util.concurrent.Flow API提供。

Reactive-Stream中核心编程模型抽象：

图2 Reactor-Steam消息处理流程

消息的处理流程如下：

当对publisher进行订阅时，publisher将Subscription对象发送到消费者，消费者适用Subscription控制消息的推送。
消费者调用Subscription.request（n）方法表示，可以消费n条消息，消息发送者消息推送至消费者直至消息数量达到n（调用消费者的onXXX方法）。
消费者持续调用Subscription.request(n)方法像消息生产者反馈自身消费能力，使得消息推送在背压得到管理的情况下持续进行。

下面是React-Stream规范定义的核心接口：

Publisher接口：一个发送无限序列的发布者。

public interface Publisher<T> {

void subscribe(Subscriber<? super T> var1);

}

Subscriber接口：定义特定事件触发的处理方法。包括onSubscribe:Publisher进行订阅时；onNext：消费消息；onError:处理异常信号；onComplete：处理结束信号。

public static interface Subscriber<T> {

public void onSubscribe(Subscription subscription);

public void onNext(T item);

public void onError(Throwable throwable);

public void onComplete();

}