文章/答案/技术大牛

发布

社区首页 >专栏 >Spring船新版推出的WebFlux，是兄弟就来学我

Spring船新版推出的WebFlux，是兄弟就来学我

端碗吹水

发布于 2020-09-23 02:06:38

2.1K00

代码可运行

文章被收录于专栏：程序猿的大杂烩程序猿的大杂烩

运行总次数：0

代码可运行

初识SpringWebFlux

Spring WebFlux是Spring Framework 5.0中引入的新的响应式Web框架。与Spring MVC不同，它不需要Servlet API，完全异步和非阻塞，并通过Reactor项目实现Reactive Streams规范，所以性能更高。并且可以在诸如Netty，Undertow和Servlet 3.1+容器的服务器上运行。注意，这里性能高指的是吞吐量方面，WebFlux 并不能使接口的响应时间缩短，它仅仅能够提升吞吐量和伸缩性。以下是官方原话：

Reactive and non-blocking generally do not make applications run faster.

所以WebFlux的目的并不是要替代Spring MVC，而是让我们在开发web应用时多一项选择。由于WebFlux实际上是对IO性能的优化，所以适合应用在IO密集型的业务场景中。

Spring WebFlux特性：

异步非阻塞: 众所周知，SpringMVC是同步阻塞的IO模型，资源浪费相对来说比较严重，当我们在处理一个比较耗时的任务时，例如：上传一个比较大的文件，首先，服务器的线程一直在等待接收文件，在这期间它就像个傻子一样等在那，什么都干不了，好不容易等到文件来了并且接收完毕，我们又要将文件写入磁盘，在这写入的过程中，这根线程又再次懵bi了，又要等到文件写完才能去干其它的事情。这一前一后的等待，不浪费资源么？没错，Spring WebFlux就是来解决这问题的，Spring WebFlux可以做到异步非阻塞。还是上面那上传文件的例子，Spring WebFlux是这样做的：线程发现文件还没准备好，就先去做其它事情，当文件准备好之后，通知这根线程来处理，当接收完毕写入磁盘的时候（根据具体情况选择是否做异步非阻塞），写入完毕后通知这根线程再来处理（异步非阻塞情况下）。相对SpringMVC而言，可以节省系统资源以及支持更高的并发量。
响应式(reactive)函数编程： Spring WebFlux支持函数式编程，得益于对于reactive-stream的支持（通过reactor框架来实现的）
不再拘束于Servlet容器：以前，我们的应用都运行于Servlet容器之中，例如我们大家最为熟悉的Tomcat, Jetty...等等。而现在Spring WebFlux不仅能运行于传统的Servlet容器中（前提是容器要支持Servlet3.1，因为非阻塞IO是使用了Servlet3.1的特性），还能运行在支持NIO的Netty和Undertow中。

Spring WebFlux与Spring MVC的对比图：

Spring WebFlux支持两种编程方式：

异步servlet

在学习webflux之前，我们首先要学习一下异步的servlet。我们需要了解同步servlet阻塞了什么？为什么需要异步servlet？异步servlet能支持高吞吐量的原理是什么？

servlet容器（如tomcat）里面，每处理一个请求会占用一个线程，同步servlet里面，业务代码处理多久，servlet容器的线程就会等（阻塞）多久，而servlet容器的线程是由上限的，当请求多了的时候servlet容器线程就会全部用完，就无法再处理请求（这个时候请求可能排队也可能丢弃，得看如何配置），就会限制了应用的吞吐量！

而异步serlvet里面，servlet容器的线程不会傻等业务代码处理完毕，而是直接返回（继续处理其他请求），给业务代码一个回调函数（asyncContext.complete()），业务代码处理完了再通知我！这样就可以使用少量的线程处理更加高的请求，从而实现高吞吐量！

我们来看一个同步Servlet的示例代码：

package org.example.servlet;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @program: servlet-demo
 * @description: 同步的Servlet Demo
 * @author: 01
 * @create: 2018-10-04 17:02
 **/
@WebServlet("/SyncServlet")
public class SyncServlet extends HttpServlet {

    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        long timeMillis = System.currentTimeMillis();

        // 执行业务代码
        doSometing(req, resp);

        System.out.println("sync use: " + (System.currentTimeMillis() - timeMillis));
    }

    private void doSometing(HttpServletRequest req, HttpServletResponse resp) throws IOException {
        // 模拟耗时操作
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        resp.getWriter().append("done");
    }
}

运行结果如下：

sync use: 5000

从运行结果可以看到，业务代码花了5 秒，但servlet容器的线程几乎没有任何耗时。而如果是同步servlet的，线程就会傻等5秒，这5秒内这个线程只处理了这一个请求。

然后我们来看一下异步Servlet的示例代码：

package org.example.servlet;

import javax.servlet.AsyncContext;
import javax.servlet.ServletException;
import javax.servlet.ServletRequest;
import javax.servlet.ServletResponse;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

/**
 * @program: servlet-demo
 * @description: 异步的Servlet Demo
 * @author: 01
 * @create: 2018-10-04 17:16
 **/
@WebServlet(asyncSupported = true, urlPatterns = "/AsyncServlet")
public class AsyncServlet extends HttpServlet {

    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        long timeMillis = System.currentTimeMillis();

        // 1.开启异步上下文
        AsyncContext asyncContext = req.startAsync();

        // 2.异步执行业务代码，放到另一个线程去处理
        CompletableFuture.runAsync(() -> doSometing(asyncContext, asyncContext.getRequest(), asyncContext.getResponse()));

        System.out.println("async use: " + (System.currentTimeMillis() - timeMillis));
    }

    private void doSometing(AsyncContext asyncContext, ServletRequest req, ServletResponse resp) {
        // 模拟耗时操作
        try {
            TimeUnit.SECONDS.sleep(5);
            resp.getWriter().append("done");
        } catch (InterruptedException | IOException e) {
            e.printStackTrace();
        }

        // 3.业务代码处理完毕，通知请求结束
        asyncContext.complete();
    }
}

运行结果如下：

async use: 8

可以看到，异步的Servlet不会阻塞Tomcat的线程，异步Servlet可以把耗时的操作交给另一个线程去处理，从而使得Tomcat的线程能够继续接收下一个请求。这就是异步Servlet的工作方式，得益于非阻塞的特性，能够大大提高服务器的吞吐量。

Webflux开发

了解了同步的Servlet和异步Servlet之间的区别以及异步Servlet的工作方式之后，我们就可以开始尝试使用一下Spring的webflux了。

创建一个Spring Boot工程，选择如下依赖：

关于reactor：

spring webflux是基于reactor来实现响应式的。那么reactor是什么呢？我是这样理解的 reactor = jdk8的stream + jdk9的flow响应式流。理解了这句话，reactor就很容易掌握。reactor里面Flux和Mono就是stream，它的最终操作就是 subscribe/block 2种。

Reactor中的Mono和Flux：

Flux 和 Mono 是 Reactor 中的两个基本概念。Flux 表示的是包含 0 到 N 个元素的异步序列。在该序列中可以包含三种不同类型的消息通知：正常的包含元素的消息、序列结束的消息和序列出错的消息。当消息通知产生时，订阅者中对应的方法 onNext(), onComplete()和 onError()会被调用。Mono 表示的是包含 0 或者 1 个元素的异步序列。该序列中同样可以包含与 Flux 相同的三种类型的消息通知。Flux 和 Mono 之间可以进行转换。对一个 Flux 序列进行计数操作，得到的结果是一个 Mono对象。把两个 Mono 序列合并在一起，得到的是一个 Flux 对象。了解更多>>

我们来看一段代码，理解一下reactor的概念：

package org.example.spring.webflux;

import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import reactor.core.publisher.Flux;

/**
 * @program: webflux
 * @description: Reactor Demo
 * @author: 01
 * @create: 2018-10-04 17:58
 **/
public class ReactorDemo {

    public static void main(String[] args) {
        // Mono 0-1个元素
        // Flux 0-N 个元素
        String[] strings = {"1", "2", "3"};

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

            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);

                // 处理完调用request再请求一个数据
                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("处理完了!");
            }

        };

        // reactor = jdk8 stream + jdk9 reactive stream
        // 这里就是jdk8的stream
        Flux.fromArray(strings).map(Integer::parseInt)
                // 最终操作，这里就是jdk9的reactive stream
                .subscribe(subscriber);
    }
}

在以上例子中，我们可以像JDK9那样实现订阅者，并且直接就可以用在reactor的subscribe方法上。调用了subscribe方法就相当于调用了stream的最终操作。有了 reactor = jdk8 stream + jdk9 reactive stream 概念后，在掌握了jdk8的stream和jkd9的flow之后，reactor也不难掌握。

如果对 jdk8 stream 和 jdk9 reactive stream不了解的话，可以参考我另外两篇文章：

了解了reactor的概念后，我们来编写一段测试代码，对比一下webflux的两种开发方式：

package org.example.spring.webflux.controller;

import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

import java.util.concurrent.TimeUnit;

/**
 * @program: webflux
 * @description: webflux demo
 * @author: 01
 * @create: 2018-10-04 17:47
 **/
@Slf4j
@RestController
public class TestController {

    /**
     * 传统的 spring mvc 开发方式
     */
    @GetMapping("/mvc")
    public String mvc() {
        long timeMillis = System.currentTimeMillis();
        log.info("mvc() start");
        String result = createStr();
        log.info("mvc() end use time {}/ms", System.currentTimeMillis() - timeMillis);

        return result;
    }

    /**
     * spring webflux 的开发方式
     */
    @GetMapping("/webflux")
    public Mono<String> webflux() {
        long timeMillis = System.currentTimeMillis();
        log.info("webflux() start");
        Mono<String> result = Mono.fromSupplier(this::createStr);
        log.info("webflux() end use time {}/ms", System.currentTimeMillis() - timeMillis);

        return result;
    }

    private String createStr() {
        // 模拟耗时操作
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "some string";
    }
}

访问/mvc，控制台输出日志如下：

访问/webflux，控制台输出日志如下：

以上的例子中，只演示了reactor 里的mono操作，返回了0-1个元素。以下示例则简单演示了flux操作，返回0-N个元素，代码如下：

/**
 * 使用flux，像流一样返回0-N个元素
 */
@GetMapping(value = "/flux", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> flux() {
    long timeMillis = System.currentTimeMillis();
    log.info("webflux() start");
    Flux<String> result = Flux.fromStream(IntStream.range(1, 5).mapToObj(i -> {
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "flux data--" + i;
    }));
    log.info("webflux() end use time {}/ms", System.currentTimeMillis() - timeMillis);

    return result;
}

访问/flux接口后，控制台输出日志如下：

在浏览器上会每隔一秒接收一行数据：

SSE（Server-Sent Events）

在上一小节的例子中我们使用flux返回数据时，可以多次返回数据（其实和响应式没有关系），实际上使用的技术就是H5的SSE。我们学习一个技术，API的使用只是最初级也是最简单的，更加重要的是需要知其然并知其所以然，否则就只能死记硬背不用就忘！我们不满足在spring里面能实现sse效果，更加需要知道spring是如何做到的。

其实SSE很简单，我们花一点点时间就可以掌握，我们在纯servlet环境里面实现。如下示例：

package org.example.servlet;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * @program: servlet-demo
 * @description: SSE Demo
 * @author: 01
 * @create: 2018-10-04 19:37
 **/
@WebServlet("/ServerSentEventsServlet")
public class ServerSentEventsServlet extends HttpServlet {

    @Override
    protected void doGet(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
        // 设置返回的数据类型及字符编码
        resp.setContentType("text/event-stream");
        resp.setCharacterEncoding("UTF-8");

        for (int i = 0; i < 5; i++) {
            // 自定义事件标识（非必须）
            resp.getWriter().write("event:me\n");

            // 需特定格式：data: + 数据 + 2个回车符
            resp.getWriter().write("data:" + i + "\n\n");
            resp.getWriter().flush();

            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

其中最为关键的是 ContentType 需为 "text/event-stream"，然后返回的数据符合固定的要求格式即可。

使用浏览器访问如下：

如果前端需要进行一些处理的话，我们也可以编写js代码来获取数据，如下示例：

<script type="text/javascript">
    // 初始化sse，参数为url
    var sse = new EventSource("ServerSentEventsServlet");

    // 无自定义事件标识时，通过onmessage事件获取返回的数据
    sse.onmessage = function (evt) {
        console.log("message", evt.data, evt)
    };

    // 若有自定义的事件标识时，通过添加事件监听获取返回的数据
    sse.addEventListener("me", function (evt) {
        console.log("message", evt.data);
        if (evt.data === 3) {
            // 关闭sse
            sse.close()
        }
    });
</script>