技术使用 Spring 5 的 WebFlux 开发反应式 Web 应用

Posted 2021-04-21 程序员交流学习

Spring 5 是流行的 Spring 框架的下一个重大的版本升级。Spring 5 中最重要改动是把反应式编程的思想应用到了框架的各个方面，Spring 5 的反应式编程以 Reactor 库为基础。Spring 5 框架所包含的内容很多，题主提到了其中新增的 WebFlux 模块。开发人员可以使用 WebFlux 创建高性能的 Web 应用和客户端。本文对 WebFlux 模块进行了详细介绍，包括其中的 HTTP、服务器推送事件和 WebSocket 支持。笔者也试着用过其中的服务器推送事件和WebSocket，要从网络上找来一些参考资料和题主分享。

WebFlux 模块的名称是 spring-webflux，名称中的 Flux 来源于 Reactor 中的类 Flux。该模块中包含了对反应式 HTTP、服务器推送事件和 WebSocket 的客户端和服务器端的支持。对于开发人员来说，比较重要的是服务器端的开发，这也是本文的重点。在服务器端，WebFlux 支持两种不同的编程模型：第一种是 Spring MVC 中使用的基于 Java 注解的方式；第二种是基于 Java 8 的 lambda 表达式的函数式编程模型。这两种编程模型只是在代码编写方式上存在不同。它们运行在同样的反应式底层架构之上，因此在运行时是相同的。WebFlux 需要底层提供运行时的支持，WebFlux 可以运行在支持 Servlet 3.1 非阻塞 IO API 的 Servlet 容器上，或是其他异步运行时环境，如 Netty 和 Undertow。

最方便的创建 WebFlux 应用的方式是使用 Spring Boot 提供的应用模板。直接访问 Spring Initializ 网站，选择创建一个 Maven 或 Gradle 项目。Spring Boot 的版本选择 2.0.0 M2。在添加的依赖中，选择 Reactive Web。最后输入应用所在的分组和名称，点击进行下载即可。需要注意的是，只有在选择了 Spring Boot 2.0.0 M2 之后，依赖中才可以选择 Reactive Web。下载完成之后可以导入到 IDE 中进行编辑。本文的示例代码使用 Intellij IDEA 2017.2 进行编写。

笔者主要从服务器推送事件和WebSocket来简单讲一下WebFlux。

服务器推送事件

服务器推送事件（Server-Sent Events，SSE）允许服务器端不断地推送数据到客户端。相对于 WebSocket 而言，服务器推送事件只支持服务器端到客户端的单向数据传递。虽然功能较弱，但优势在于 SSE 在已有的 HTTP 协议上使用简单易懂的文本格式来表示传输的数据。作为 W3C 的推荐规范，SSE 在浏览器端的支持也比较广泛，除了 IE 之外的其他浏览器都提供了支持。在 IE 上也可以使用 polyfill 库来提供支持。在服务器端来说，SSE 是一个不断产生新数据的流，非常适合于用反应式流来表示。在 WebFlux 中创建 SSE 的服务器端是非常简单的。只需要返回的对象的类型是 Flux<ServerSentEvent>，就会被自动按照 SSE 规范要求的格式来发送响应。

代码清单1中的 SseController 是一个使用 SSE 的控制器的示例。其中的方法 randomNumbers()表示的是每隔一秒产生一个随机数的 SSE 端点。我们可以使用类 ServerSentEvent.Builder 来创建 ServerSentEvent 对象。这里我们指定了事件名称 random，以及每个事件的标识符和数据。事件的标识符是一个递增的整数，而数据则是产生的随机数。

清单1 服务器推送事件示例

@RestController

@RequestMapping("/sse")

public class SseController {

    @GetMapping("/randomNumbers")

    public Flux<ServerSentEvent<Integer>> randomNumbers() {

        return Flux.interval(Duration.ofSeconds(1))

                .map(seq -> Tuples.of(seq, ThreadLocalRandom.current().nextInt()))

                .map(data -> ServerSentEvent.<Integer>builder()

                        .event("random")

                        .id(Long.toString(data.getT1()))

                        .data(data.getT2())

                        .build());

    }

}

清单2  SSE 服务器端发送的响应

id:0

event:random

data:751025203

 

id:1

event:random

data:-1591883873

 

id:2

event:random

data:-1899224227

WebSocket

WebSocket 支持客户端与服务器端的双向通讯。当客户端与服务器端之间的交互方式比较复杂时，可以使用 WebSocket。WebSocket 在主流的浏览器上都得到了支持。WebFlux 也对创建 WebSocket 服务器端提供了支持。在服务器端，我们需要实现接口 org.springframework.web.reactive.socket.WebSocketHandler 来处理 WebSocket 通讯。接口 WebSocketHandler 的方法 handle 的参数是接口 WebSocketSession 的对象，可以用来获取客户端信息、接送消息和发送消息。代码清单3中的 EchoHandler 对于每个接收的消息，会发送一个添加了"ECHO -> "前缀的响应消息。WebSocketSession 的 receive 方法的返回值是一个 Flux<WebSocketMessage>对象，表示的是接收到的消息流。而 send 方法的参数是一个 Publisher<WebSocketMessage>对象，表示要发送的消息流。在 handle 方法，使用 map 操作对 receive 方法得到的 Flux<WebSocketMessage>中包含的消息继续处理，然后直接由 send 方法来发送。

清单3  WebSocket 的 EchoHandler 示例

@Component

public class EchoHandler implements WebSocketHandler {

    @Override

    public Mono<Void> handle(final WebSocketSession session) {

        return session.send(

                session.receive()

                        .map(msg -> session.textMessage("ECHO -> " + msg.getPayloadAsText())));

    }

}

在创建了 WebSocket 的处理器 EchoHandler 之后，下一步需要把它注册到 WebFlux 中。我们首先需要创建一个类 WebSocketHandlerAdapter 的对象，该对象负责把 WebSocketHandler 关联到 WebFlux 中。