背景

I/O阻塞浪费系统性能，因此为了应对 高并发环境下 的服务端编程，微软提出了一个实现 异步编程 的方案 - Reactive Programming，即响应式编程，通过异步非阻塞处理发挥系统的性能，提升系统的吞吐量；各个语言也跟进推出了自己的响应式框架，而Reactor做为java重要的一款响应式框架也就此诞生。

Reactive Streams、Reactor和WebFlux

1. Reactive Streams 是一套响应式编程标准和规范；
2. Reactor 是基于 Reactive Streams 一套反应式编程框架；
3. WebFlux 以 Reactor 为基础，实现 Web 领域的反应式编程框架。

从上面看的reactor是基础，所以理解reactor的核心思想十分重要。Reactor 框架主要有两个主要的模块： reactor-core reactor-ipc

前者主要负责 Reactive Programming 相关的核心API 实现，后者负责 高性能网络通信 的实现。其实，对于业务开发人员来说，当编写响应式代码时，通常只会接触到 Publisher 这个接口，对应到 Reactor 便是 Mono 和 Flux。本文主要通过一个小例子给大家展示下reactor-core的原理。

代码

reactor-core中有三个重要的接口，分别表示发布者、订阅者，还有订阅关系。

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

public interface Subscriber<T> {
    void onSubscribe(Subscription var1);

    void onNext(T var1);

    void onError(Throwable var1);

    void onComplete();
}

public interface Subscription {
    void request(long var1);

    void cancel();
}

整个工作流程是这样的。

1. 发布者调用subscribe，该方法接受一个订阅者作为参数，在这个方法里面，发布者会创建一个订阅关系。该订阅关系会维持对应的订阅者从发布者获取数据的状态。
2. 发布者调用订阅者提供的onSubscribe，该方法会接受上面的订阅关系为参数
3. 订阅者的onSubscribe方法，会调用订阅关系的request方法，该方法会从发布者获取数据，然后逐一的调用订阅者的onNext方法处理获取的元素，所有元素都获得了，便调用onComplete, 当元素为空时，调用onError方法。

然后我们看一个小例子

    public static void main(String[] args) {
      Flux.just("red", "white", "blue")
            .log()
            .map(String::toUpperCase)
            .subscribe(System.out::println);
   }

上面的整个过程第2行调用Flux.just会创建一个FluxArray，FluxArray是一种Flux，

然后第4行调用map方法会创建一个FluxMapFuseable，它也是个Flux.同时它保存了FluxArray的引用。

然后第5行调用subscribe方法，它会创建LamdaSubscriber，同时会调用FluxMapFuseable的subscribeOrReturn,获得一个CoreSubscriber(它保存了map的操作),此时如下面的第三个图，然后调用它的OnNext方法把每个元素变成大写。然后又调用LamdaSubscriber的OnNext方法进行元素输出。

Tags：Java编程思想

微信扫一扫：分享

微信里点“发现”，扫一下

二维码便可将本文分享至朋友圈。