1.反應式編程簡介

反應式編程是為具有以下特征的應用程序創造的術語：

• 非阻塞應用
• 事件驅動和異步
• 需要少量線程來垂直擴展（即在JVM中）

就像面向對象的編程，函數式編程或過程式編程一樣，反應式編程只是另一種編程范例。 它使我們的程序成為：響應式，彈性，彈性。

2. Spring的反應式編程

Spring框架內部使用Reactor為其自身提供響應式支持。 Reactor是Reactive Streams（發布程序，在Java9中引入）的實現。 Reactor具有以下兩種數據類型：

• 助焊劑（它是一個可以發射0個或多個元素的流）
• 單聲道（它是一個可以發出0或1個元素的流）

Spring從其API中公開這些類型，從而使應用程序具有響應性。

在Spring5中，引入了一個名為WebFlux的新模塊，該模塊支持使用以下方式創建反應式Web應用程序：HTTP（REST）和Web套接字。

Spring Web Flux支持以下兩種模型：

• 功能模型
• 注釋模型

在本文中，我們將探討功能模型。

下表比較了普通Spring和Web Flux：

傳統堆棧	反應堆
Spring Web MVC	SpringWebFlux
控制器和處理程序映射	路由器功能
Servlet API	HTTP /反應流
Servlet容器	任何支持Servlet 3.1 +，Tomcat 8.x，Jetty，Netty，UnderTow的servlet容器

3.用例

必須為員工管理系統創建一個使用Spring Web Flux的REST API，該系統將在員工身上公開CRUD。

注意：項目的DAO層是硬編碼的。

4.所需的軟件和環境

• Java：1.8以上
• Maven：3.3.9或更高
• Eclipse Luna或以上
• Spring靴：2.0.0.M4
• Spring Boot Starter WebFlux
• 郵遞員測試應用程序

5.申請流程

Spring5 WebFlux的功能模型是使用Spring MVC樣式注釋的替代方法。 在Spring WebFlux功能模型中，路由器和處理程序函數用于創建MVC應用程序.HTTP請求通過路由器函數路由（替代@RequestMapping注解），請求通過處理程序函數 （替代@Controller處理程序方法）進行處理。

每個處理程序函數都將ServerRequest （ org.springframework.web.reactive.function.server.ServerRequest ）作為參數，結果將返回Mono<ServerResponse>或Flux<ServerResponse> （ org.springframework.web.reactive.function.server.ServerResponse ）。

6.用例代碼和描述

pom.xml

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.webflux</groupId><artifactId>Demo_Spring_MVC_Web_Flux</artifactId><version>0.0.1-SNAPSHOT</version><repositories><repository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><snapshots><enabled>true</enabled></snapshots></repository><repository><id>spring-milestones</id><name>Spring Milestones</name><url>https://repo.spring.io/milestone</url><snapshots><enabled>false</enabled></snapshots></repository></repositories><pluginRepositories><pluginRepository><id>spring-snapshots</id><name>Spring Snapshots</name><url>https://repo.spring.io/snapshot</url><snapshots><enabled>true</enabled></snapshots></pluginRepository><pluginRepository><id>spring-milestones</id><name>Spring Milestones</name><url>https://repo.spring.io/milestone</url><snapshots><enabled>false</enabled></snapshots></pluginRepository></pluginRepositories><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.0.0.M4</version><relativePath /><!-- lookup parent from repository --></parent><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding><!-- Configuring Java 8 for the Project --><java.version>1.8</java.version></properties><!--Excluding Embedded tomcat to make use of the Netty Server--><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId><exclusions><exclusion><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-tomcat</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build> </project>

EmployeeDAO.java

package com.webflux.dao; import java.util.LinkedHashMap; import java.util.Map; import org.springframework.stereotype.Repository; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import com.webflux.bussiness.bean.Employee; @Repository public class EmployeeDAO {/*** Map is used to Replace the Database * */static public Map mapOfEmloyeess = new LinkedHashMap();static int count=10004;static{mapOfEmloyeess.put(10001, new Employee("Jack",10001,12345.6,1001));mapOfEmloyeess.put(10002, new Employee("Justin",10002,12355.6,1002));mapOfEmloyeess.put(10003, new Employee("Eric",10003,12445.6,1003));}/*** Returns all the Existing Employees as Flux* */public Flux getAllEmployee(){return Flux.fromStream(mapOfEmloyeess.values().stream()); }/**Get Employee details using EmployeeId .* Returns a Mono response with Data if Employee is Found* Else returns a null* */public Mono getEmployeeDetailsById(int id){Monores = null;Employee emp =mapOfEmloyeess.get(id);if(emp!=null){res=Mono.just(emp);}return res;}/**Create Employee details.* Returns a Mono response with auto-generated Id* */public Mono addEmployee(Employee employee){count++;employee.setEmployeeId(count);mapOfEmloyeess.put(count, employee);return Mono.just(count);}/**Update the Employee details,* Receives the Employee Object and returns the updated Details* as Mono * */public Mono updateEmployee (Employee employee){mapOfEmloyeess.put(employee.getEmployeeId(), employee);return Mono.just(employee);}/**Delete the Employee details,* Receives the EmployeeID and returns the deleted employee Details* as Mono * */public Mono removeEmployee (int id){Employee emp= mapOfEmloyeess.remove(id);return Mono.just(emp);} }

可以觀察到， EmployeeDAO所有方法都返回Mono或Flux Response，從而使DAO層處于活動狀態。

EmployeeHandler.java

package com.webflux.web.handler; import static org.springframework.web.reactive.function.BodyInserters.fromObject; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.MediaType; import org.springframework.stereotype.Controller; import org.springframework.web.reactive.function.BodyExtractors; import org.springframework.web.reactive.function.server.ServerRequest; import org.springframework.web.reactive.function.server.ServerResponse; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import com.webflux.bussiness.bean.Employee; import com.webflux.dao.EmployeeDAO; @Controller public class EmployeeHandler {@Autowiredprivate EmployeeDAO employeeDAO;/*** Receives a ServerRequest.* Invokes the method getAllEmployee() from EmployeeDAO.* Prepares a Mono and returns the same.* */ public Mono getEmployeeDetails(ServerRequest request) {Flux res=employeeDAO.getAllEmployee(); return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(res,Employee.class);}/*** Receives a ServerRequest.* Extracts the Path Variable (named id) from the Request.* Invokes the method [getEmployeeDetailsById()] from EmployeeDAO.* Verifies if the object returned in the previous step is null * then returns a Bad request with appropriate message.* Else Returns the Mono with the Employee Data.* */public Mono getEmployeeDetailByEmployeeId(ServerRequest request) {//Extracts the Path Variable id from the Requestint id =Integer.parseInt(request.pathVariable("id"));Mono employee = employeeDAO.getEmployeeDetailsById(id);Mono res= null;if(employee==null){res=ServerResponse.badRequest().body(fromObject("Please give a valid employee Id"));}else{//Converting Mono of Mono type to Monores=employee.flatMap(x->ServerResponse.ok().body(fromObject(x))); }return res;}/*** Receives a ServerRequest.* Makes use of BodyExtractors and Extracts the Employee Data as * Mono from the ServerRequest.* Invokes the method [addEmployee()] of the EmployeeDAO.* Prepares a Mono and returns the same. * */public Mono addEmployee(ServerRequest request) {Mono requestBodyMono = request.body(BodyExtractors.toMono(Employee.class));Mono mono= employeeDAO.addEmployee(requestBodyMono.block());//Converting Mono of Mono type to MonoMono res= mono.flatMap(x->ServerResponse.ok().body(fromObject("Employee Created with Id"+x))); return res;}/*** Receives a ServerRequest.* Makes use of BodyExtractors and Extracts the Employee Data as * Mono from the ServerRequest.* Finds the Employee and updates the details by invoking updateEmployee() of * EmployeeDAO. * Prepares a Mono and returns the same.* */public Mono updateEmployee(ServerRequest request) {Mono requestBodyMono = request.body(BodyExtractors.toMono(Employee.class));Employee employee = requestBodyMono.block();Mono employeeRet = employeeDAO.getEmployeeDetailsById(employee.getEmployeeId());Mono res= null;if(employeeRet==null){res=ServerResponse.badRequest().body(fromObject("Please Give valid employee details to update"));}else{Mono emp= employeeDAO.updateEmployee(employee);//Converting Mono of Mono type to Monores=emp.flatMap(x->ServerResponse.ok().body(fromObject(x)));}return res; }/*** Receives a ServerRequest.* Makes use of BodyExtractors and Extracts the Employee Data as * Mono from the ServerRequest.* Finds the Employee and deletes the details by invoking removeEmployee() of * EmployeeDAO. * Prepares a Mono and returns the same.* */public Mono deleteEmployee(ServerRequest request) {int myId = Integer.parseInt(request.pathVariable("id"));Mono res= null;if (employeeDAO.getEmployeeDetailsById(myId) == null) {res=ServerResponse.badRequest().body(fromObject("Please Give valid employee details to delete"));}else{Mono employee = employeeDAO.removeEmployee(myId);//Converting Mono of Mono type to Monores=employee.flatMap(x->ServerResponse.ok().body(fromObject(x))); }return res;} }

可以看出，Handler的所有方法都返回Mono<ServerResponse> ，從而使表示層處于響應狀態。

注意 ：事件處理程序方法應接受ServerRequest并應返回Mono<ServerResponse>

RouterConfiguration.java

package com.webflux.web.router.config; import static org.springframework.web.reactive.function.server.RequestPredicates.DELETE; import static org.springframework.web.reactive.function.server.RequestPredicates.GET; import static org.springframework.web.reactive.function.server.RequestPredicates.POST; import static org.springframework.web.reactive.function.server.RequestPredicates.PUT; import static org.springframework.web.reactive.function.server.RequestPredicates.accept; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.http.MediaType; import org.springframework.web.reactive.function.server.RouterFunction; import org.springframework.web.reactive.function.server.RouterFunctions; import org.springframework.web.reactive.function.server.ServerResponse; import com.webflux.web.handler.EmployeeHandler; @Configuration /*** Router is configuration class.* It links the incoming requests with appropriate HTTP methods to the * respective method of the EmployeeHandler.* Method references are used for the mapping.* */ public class RouterConfiguration{@AutowiredEmployeeHandler employeeHandler;@Beanpublic RouterFunction monoRouterFunction() {RouterFunctionrouterFunction= RouterFunctions. route(GET("/emp/controller/getDetails").and(accept(MediaType.APPLICATION_JSON)), employeeHandler::getEmployeeDetails).andRoute(GET("/emp/controller/getDetailsById/{id}").and(accept(MediaType.APPLICATION_JSON)), employeeHandler::getEmployeeDetailByEmployeeId).andRoute(POST("/emp/controller/addEmp").and(accept(MediaType.APPLICATION_JSON)), employeeHandler::addEmployee).andRoute(PUT("/emp/controller/updateEmp").and(accept(MediaType.APPLICATION_JSON)), employeeHandler::updateEmployee).andRoute(DELETE("/emp/controller/deleteEmp/{id}").and(accept(MediaType.APPLICATION_JSON)), employeeHandler::deleteEmployee);return routerFunction;} }

ApplicationBootUp.java

package com.webflux; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class ApplicationBootUp {public static void main(String[] args) {SpringApplication.run(ApplicationBootUp.class);}}

在application.properties內部，僅提及服務器端口： server.port=8090 。

可以使用以下命令部署應用程序： clean install spring-boot:run并使用postman client進行測試。

7.參考：

• https://docs.spring.io/spring/docs/5.0.0.BUILD-SNAPSHOT/spring-framework-reference/html/web-reactive.html
• http://www.baeldung.com/reactor-core
• http://www.baeldung.com/spring-5-functional-web

8.下載Eclipse項目

下載
您可以在此處下載此示例的完整源代碼： SpringWebFlux

翻譯自: https://www.javacodegeeks.com/2018/01/reactive-web-applications-using-springwebflux.html

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的使用SpringWebFlux的反应式Web应用程序的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。