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

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

在测试 SSE 时，咱们只须要使用 curl 来访问便可。代码清单2 给出了调用某地址的结果。
清单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 中。