WebFlux前置知识（四）

[TOC]

# 1.Backpressure

Backpressure 在国内被翻译成背压，这个翻译在网上被很多人吐槽，我觉得大家的吐槽是有道理的，背压单纯从字面上确实看不出来有什么意思。所以松哥这里直接用英文 Backpressure 吧。

Backpressure 是一种现象：当数据流从上游生产者向下游消费者传输的过程中，上游生产速度大于下游消费速度，导致下游的 Buffer 溢出，这种现象就叫做 Backpressure。

换句话说，上游生产数据，生产完成后通过管道将数据传到下游，下游消费数据，当下游消费速度小于上游数据生产速度时，数据在管道中积压会对上游形成一个压力，这就是 Backpressure，从这个角度来说，Backpressure 翻译成反压、回压似乎更合理一些。

Backpressure 会出现在有 Buffer 上限的系统中，当出现 Buffer 溢出的时候，就会有 Backpressure，对于 Backpressure，它的应对措施只有一个：丢弃新事件。那么什么是 Buffer 溢出呢？例如我的服务器可以同时处理 2000 个用户请求，那么我就把请求上限设置为 2000，这个 2000 就是我的 Buffer，当超出 2000 的时候，就产生了 Backpressure。

# 2.Flow API

JDK9 中推出了 Flow API，用以支持 Reactive Programming，即响应式编程。

在响应式编程中，会有一个数据发布者 Publisher 和数据订阅者 Subscriber，Subscriber 接收 Publisher 发布的数据并进行消费，在 Subscriber 和 Publisher 之间还存在一个 Processor，类似于一个过滤器，可以对数据进行中间处理。

JDK9 中提供了 Flow API 用以支持响应式编程，另外 RxJava 和 Reactor 等框架也提供了相关的实现。

我们来看看 JDK9 中的 Flow 类：

非常简洁，基本上就是按照 Reactive Programming 的设计来的：

Publisher

Publisher 为数据发布者，这是一个函数式接口，里边只有一个方法，通过这个方法将数据发布出去，Publisher 的定义如下：

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

Subscriber

Subscriber 为数据订阅者，这个里边有四个方法，如下：

public static interface Subscriber<T> {
    public void onSubscribe(Subscription subscription);
    public void onNext(T item);
    public void onError(Throwable throwable);
    public void onComplete();
}

onSubscribe：这个是订阅成功的回调方法，用于初始化 Subscription，并且表明可以开始接收订阅数据了。
onNext：接收下一项订阅数据的回调方法。
onError：在 Publisher 或 Subcriber 遇到不可恢复的错误时调用此方法，之后 Subscription 不会再调用 Subscriber 其他的方法。
onComplete：当接收完所有订阅数据，并且发布者已经关闭后会回调这个方法。

Subscription

Subscription 为发布者和订阅者之间的订阅关系，用来控制消息的消费，这个里边有两个方法：

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

request：这个方法用来向数据发布者请求 n 个数据。
cancel：取消消息订阅，订阅者将不再接收数据。

Processor

Processor 是一个空接口，不过它同时继承了 Publisher 和 Subscriber，所以它既能发布数据也能订阅数据，因此我们可以通过 Processor 来完成一些数据转换的功能，先接收数据进行处理，处理完成后再将数据发布出去，这个也有点类似于我们 JavaEE 中的过滤器。

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

# 2.1 消息订阅初体验

我们通过如下一段代码体验一下消息的订阅与发布：

public class FlowDemo {
    public static void main(String[] args) {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;
            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                //向数据发布者请求一个数据
                this.subscription.request(1);
            }
            @Override
            public void onNext(String item) {
                System.out.println("接收到 publisher 发来的消息了：" + item);
                //接收完成后，可以继续接收或者不接收
                //this.subscription.cancel();
                this.subscription.request(1);
            }
            @Override
            public void onError(Throwable throwable) {
                //出现异常，就会来到这个方法，此时直接取消订阅即可
                this.subscription.cancel();
            }
            @Override
            public void onComplete() {
                //发布者的所有数据都被接收，并且发布者已经关闭
                System.out.println("数据接收完毕");
            }
        };
        //配置发布者和订阅者
        publisher.subscribe(subscriber);
        for (int i = 0; i < 5; i++) {
            //发送数据
            publisher.submit("hello:" + i);
        }
        //关闭发布者
        publisher.close();
        new Scanner(System.in).next();
    }
}

松哥稍微解释一下上面这段代码：

首先创建一个 SubmissionPublisher 对象作为消息发布者。
接下来创建 Flow.Subscriber 对象作为消息订阅者，实现消息订阅者里边的四个方法，分别进行处理。
为 publisher 配置上 subscriber。
发送消息。
消息发送完成后关闭 publisher。
最后是让程序不要停止，观察消息订阅者打印情况。

# 2.2 模拟 Backpressure

Backpressure 问题在 Flow API 中得到了很好的解决。Subscriber 会将 Publisher 发布的数据缓存在 Subscription 中，其长度默认为256，相关源码如下：

public final class Flow {
    static final int DEFAULT_BUFFER_SIZE = 256;
    public static int defaultBufferSize() {
        return DEFAULT_BUFFER_SIZE;
    }
    ...
}

一旦超出这个数据量，publisher 就会降低数据发送速度。

我们对上面的案例进行修改，如下：

public class FlowDemo {
    public static void main(String[] args) {
        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();

        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                //向数据发布者请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("接收到 publisher 发来的消息了：" + item);
                //接收完成后，可以继续接收或者不接收
                //this.subscription.cancel();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                //出现异常，就会来到这个方法，此时直接取消订阅即可
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                //发布者的所有数据都被接收，并且发布者已经关闭
                System.out.println("数据接收完毕");
            }
        };
        publisher.subscribe(subscriber);
        for (int i = 0; i < 500; i++) {
            System.out.println("i--------->" + i);
            publisher.submit("hello:" + i);
        }
        //关闭发布者
        publisher.close();
        new Scanner(System.in).next();
    }
}

一共修改了三个地方：

Subscriber#onNext 方法中，每次休息两秒再处理下一条数据。
发布数据时，一共发布 500 条数据。
打印数据发布的日志。

修改完成后，我们再次启动项目，观察控制台输出：

可以看到，生产者先是一股脑生产了 257 条数据（hello0 在一开始就被消费了，所以缓存中实际上是 256 条），消息则是一条一条的来，由于消费的速度比较慢，所以当缓存中的数据超过 256 条之后，接下来都是消费一条，再发送一条。

# 2.3 数据处理

Flow.Processor 可以像过滤器一样，对数据进行预处理，数据从 publisher 出来之后，先进入 Flow.Processor 中进行预处理，然后再进入 Subscriber。

修改后的代码如下：

public class FlowDemo {
    public static void main(String[] args) {

        class DataFilter extends SubmissionPublisher<String> implements Flow.Processor<String,String>{

            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                this.submit("【这是一条被处理过的数据】" + item);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                this.close();
            }
        }

        SubmissionPublisher<String> publisher = new SubmissionPublisher<>();
        DataFilter dataFilter = new DataFilter();
        publisher.subscribe(dataFilter);

        Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
            private Flow.Subscription subscription;

            @Override
            public void onSubscribe(Flow.Subscription subscription) {
                this.subscription = subscription;
                //向数据发布者请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(String item) {
                System.out.println("接收到 publisher 发来的消息了：" + item);
                //接收完成后，可以继续接收或者不接收
                //this.subscription.cancel();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                //出现异常，就会来到这个方法，此时直接取消订阅即可
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                //发布者的所有数据都被接收，并且发布者已经关闭
                System.out.println("数据接收完毕");
            }
        };
        dataFilter.subscribe(subscriber);
        for (int i = 0; i < 500; i++) {
            System.out.println("发送消息 i--------->" + i);
            publisher.submit("hello:" + i);
        }
        //关闭发布者
        publisher.close();
        new Scanner(System.in).next();
    }
}

简单起见，我这里创建了一个局部内部类 DataFilter，DataFilter 继承自 SubmissionPublisher 并实现了 Flow.Processor 接口，由于 DataFilter 继承自 SubmissionPublisher，所以它也兼具 SubmissionPublisher 的功能。

在 DataFilter 中完成消息的处理并重新发送出去。接下来定义 publisher，让 dataFilter 作为其订阅者，再定义新的订阅者，作为 dataFilter 的订阅者。

最终运行效果如下：

# 3.小结

好啦，这就是今天和大家介绍的 Java9 中的 Reactive Stream，那么至此，我们的 WebFlux 前置知识差不多告一段落了，下篇文章开始，正式开整 WebFlux。

← WebFlux前置知识（三）异步 Servlet →