Spring WebFlux学习记录

Posted 2021-11-07 AlaGeek

本文按照如下顺序一步步深入解释WebFlux是个什么东西：

1、Reactive Stream
2、Reactor
3、WebFlux

其中 Reactive Stream 是 Java 9 新增的一个重要特性，而 Reactor 就相当于 Java 8 的 Stream 流 + Java 9 的 Reactive Stream，最后的 WebFlux 的核心就是基于 Reactor 的相关 API 开发的。

---

1、Reactive Stream

响应式流是 Java 9 引入的一套基于发布/订阅模式的数据处理规范，它的目标是以非阻塞背压方式实现数据的异步流。

这个是网上摘抄的一个定义，其中背压原先是一个工程里的概念，指的是在管道运输中，气流或液流由于管道突然变细、急弯等原因导致由某处出现了下游向上游的逆向压力，而在响应式流中就被引申为了：当数据流传输过程中，发布者生产数据的速度大于订阅者消费数据的速度时，订阅者告诉发布者暂时不需要更多的数据了这么一个交互动作。实际上就是流控。

Java 9 的响应式流精简为了四个接口：

• Publisher
• Subscriber
• Subscription
• Processor

其中 Publisher 是发布者接口，Subscriber 是订阅者接口，Subscription 接口用于联系发布者和订阅者，你可以理解为“合同”，两者签订了合同后，才能进行数据流传输，Processor 接口既是发布者又是订阅者，用于中间数据处理，承上启下。

以下是用这四个接口写的一个案例：

public class FlowDemo {

    public static void main(String[] args) throws InterruptedException {

        // 定义发布者
        SubmissionPublisher<Integer> publisher = new SubmissionPublisher<>();

        MyProcessor processor = new MyProcessor();

        publisher.subscribe(processor);

        // 定义订阅者
        Flow.Subscriber<String> subscriber = new Flow.Subscriber<>() {

            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("subscriber 接收数据: " + item);
                // 再请求一个数据
                this.subscription.request(1);
                // 达到目标，调用cancel告诉发布者不再接收数据
                //this.subscription.cancel();
            }

            @Override
            public void onError(Throwable throwable) {
                // 出现异常
                throwable.printStackTrace();
                // 停止接收数据
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("subscriber 处理完毕");
            }
        };

        // 发布者和订阅者建立订阅关系
        processor.subscribe(subscriber);

        // 生产数据，并发布
        for (int i = 0, j = 1; i < 4; i++, j *= -1) {
            System.out.println("publisher 生成数据: " + i * j);
            publisher.submit(i * j);
        }

        // 关闭发布者
        publisher.close();

        // 主线程延时
        Thread.currentThread().join(10000);

        // 关闭中转
        processor.close();
    }

}

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

    private Flow.Subscription subscription;

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

    @Override
    public void onNext(Integer item) {
        System.out.println("processor 接收数据: " + item);
        item = item < 0 ? -item : item;
        this.submit(item + "（转换后）");
        this.subscription.request(1);
    }

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

    @Override
    public void onComplete() {
        System.out.println("processor 中转完成");
    }
}

这个案例中用 Publisher 的实现类 SubmissionPublisher 定义了一个发布者，用 Subscriber 定义了一个订阅者，其中实现的方法中，onSubscribe 用于和发布者建立联系，onNext 用于接收数据，onError 用于异常处理，onComplete 在发布者关闭后执行，中转者通过继承 SubmissionPublisher 以及实现接口 Processor 来执行发布者和订阅者的职责，如下是案例执行结果：

发布者与中转者建立订阅关系，中转者与订阅者建立订阅关系，发布者将循环生成的数据发送给中转者，中转者将负数全部处理为正数，然后发送给订阅者。

2、Reactor

Reactor与Spring是兄弟项目，侧重于Server端的响应式编程，是一个基于 Java 8 的实现了响应式流规范的响应式库，虽然说是基于 Java 8 实现的，但是简单来说 Reactor 可以理解为是 Java 8 stream + Java 9 reactive stream。

在 Reactor 中发布者由 Flux 和 Mono 两个类定义：一个 Flux 对象代表了一个包含0-N个元素的序列，而一个 Mono 对象代表了一个包含0-1个元素的序列。

代码样例如下：

public class ReactorDemo {

    public static void main(String[] args) {

        // 定义订阅者
        Subscriber<Integer> subscriber = new Subscriber<Integer>() {

            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                // 保存订阅关系，需要用它来给发布者响应
                this.subscription = subscription;
                // 请求一个数据
                this.subscription.request(1);
            }

            @Override
            public void onNext(Integer item) {
                // 接收数据并处理
                System.out.println("接收到数据: " + item);
                // 再请求一个数据
                this.subscription.request(1);
                // 达到目标，调用cancel告诉发布者不再接收数据
                //this.subscription.cancel();
            }

            @Override
            public void onError(Throwable throwable) {
                // 出现异常
                throwable.printStackTrace();
                // 停止接收数据
                this.subscription.cancel();
            }

            @Override
            public void onComplete() {
                System.out.println("处理完毕");
            }
        };

        String[] strings = {"1","2","3"};

        Flux.fromArray(strings)
                // jdk8 stream
                .map(Integer::parseInt)
                // jdk9 reactive stream
                .subscribe(subscriber);
    }

}

代码中的订阅者实际是从第一步中的代码拷贝过来的，可以看到两个代码在订阅者的包引入上有所差异，原因是第一步中的案例是基于 Java 9 开发的，而第二步中的案例是基于 Java 8 + Reactor相关jar包 开发的。

代码中首先用 Flux 的 fromArray 方法将字符串数组转化为了一个数据流，然后因为订阅者消费的数据是Integer类型，所以用 map 将 String 转为 Integer，最后调用 subscribe 方法对流进行消费，这个方法的传参可以传入一个订阅者。

样例执行结果如下：

3、Spring WebFlux

讲 Reactor 主要是为了引出 Flux 和 Mono，实际上这部分看懂已经足够学习WebFLux（没看懂可能也能学）。

Spring WebFlux 是随着 Spring5 推出的响应式编程框架，与 Spring MVC 相比，WebFlux 最大的特点就是非阻塞，这部分直接有代码来讲解。

3.1 类MVC开发模式

首先创建一个测试项目，在pom文件中添加如下依赖：

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

然后创建一个controller，写入以下测试代码：

@GetMapping("/1")
public String get1() {
    long start = System.currentTimeMillis();
    log.info("get1 start");
    String str = createStr("1");
    log.info("get1 end");
    long end = System.currentTimeMillis();
    log.info("get1 use time: {}", end - start);
    return str;
}

@GetMapping("/2")
public Mono<String> get2() {
    long start = System.currentTimeMillis();
    log.info("get2 start");
    Mono<String> mono = Mono.fromSupplier(() -> createStr("2"));
    log.info("get2 end");
    long end = System.currentTimeMillis();
    log.info("get2 use time: {}", end - start);
    return mono;
}

private String createStr(String str) {
    try {
        TimeUnit.SECONDS.sleep(5);
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return str;
}

其中 get1 方法是典型的 MVC 写法，在方法中调用 createStr 方法，为了模拟阻塞，在 createStr 方法中加了5秒延时，而 get2 方法就是 WebFlux 的写法，在方法体中通过语句 Mono.fromSupplier(() -> createStr(“2”)) 来构造一个数据流，这里需要注意的是，流是具有惰性处理的特性的，如果方法体内没有消费这个流，那么实际上 createStr是不会执行的，那么阻塞的5秒也就不存在了。如下是代码执行结果：

可以看到，get2方法确实没有产生延迟，因为它只是返回了一个流，而这个流是在方法体执行完后，被 Spring WebFlux 框架消费的，以运行在 Tomcat 上为例，这样的写法就不会导致 Servlet线程 处于堵塞状态，从而提升吞吐量。

上面展示的是返回值为 Mono 的情况，下面演示种返回值为 Flux 的案例：

@GetMapping(value = "/3", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> flux() {
    long start = System.currentTimeMillis();
    log.info("get3 start");
    Flux<String> stringFlux = Flux.fromStream(IntStream.range(1, 5).mapToObj(
        i -> {
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "flux data " + i + "\\n";
        }
    ));
    log.info("get3 end");
    long end = System.currentTimeMillis();
    log.info("get3 use time: {}", end - start);
    return stringFlux;
}

Flux 和 Mono最大的区别就在于 Flux 产生的数据流可能会有多个元素，那么对于返回给调用方就有两种方式，一种是所有元素都消费完一起返回，另一种是消费一个返回一个，上述代码中的 MediaType.TEXT_EVENT_STREAM_VALUE 就是告诉浏览器一个个返回，如下是运行结果：
第一次返回：

第四次返回：

同样，方法体内消耗的时间如下：

3.2 Router Function开发模式

上面代码的开发模式其实跟Spring MVC的差别不大，也是写一个controller，里面的方法用@RequestMapping注解，而WebFlux还有一种开发，叫做 Router Function。

对应controller中的方法，Router Function开发模式有 HandlerFunction 类：

Mono<T extends ServerResponse> handle(ServerRequest request);

而请求路由由类 RouterFunction 实现：

Mono<HandlerFunction<T>> route(ServerRequest request);

就比如将上面的 get2 方法改写如下：

@Component
public class TestHandler {

    public Mono<ServerResponse> get2(ServerRequest serverRequest) {
        return ok().contentType(MediaType.TEXT_PLAIN).body(Mono.fromSupplier(() -> createStr("2")), String.class);
    }

    private String createStr(String str) {
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return str;
    }

}

@Configuration
public class RouterConfig {

    @Resource
    private TestHandler testHandler;

    @Bean
    public RouterFunction<ServerResponse> timerRouter() {
        return route(GET("/get2"), testHandler::get2);
    }
}

启动服务后，访问 /get 就会调用 testHandler 的 get2 方法来处理请求。

---

参考文献：

1、https://blog.csdn.net/get_set/article/details/79466657

2、https://www.bilibili.com/video/BV1y44y117pp?share_source=copy_web