使用 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 中。