20 天前

WebFlux

15 分钟读完 (大约 2298 个字)

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 的订阅者。

最终运行效果如下：

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

# WebFlux

喜欢这篇文章吗？扫码关注公众号【江南一点雨】，【江南一点雨】专注于 SPRING BOOT+微服务以及前后端分离技术，每天推送原创技术干货，关注后回复 JAVA，领取松哥为你精心准备的 JAVA 干货!

WebFlux 前置知识（四）

WebFlux 前置知识（四）

1.Backpressure

2.Flow API

2.1 消息订阅初体验

2.2 模拟 Backpressure

2.3 数据处理

Recommend

谁将接纳中国的比特币矿工？哪里会是他们新的归宿和起点？

How to repair the distance from the range of column arrays in HighCharts?

[zz]为什么古代汉族人不能在蒙古草原游牧？

GitHub - shmykelsa/AAAD

WebFlux 前置知识（三）

RewriteRule / RewriteCond help - matches anything except a directory

Reactive background image in full width div

WebFlux 中的请求地址路由怎么玩？

UK Adults Actively Seek Crypto Exposure In Search of "Viable Investments&qu...

Break a Sweat and Score Big with the 1xBit EURO 2020 Tournament

About Joyk