Spring WebFlux 教程 · HowToDoInJava 中文系列教程

# Spring WebFlux 教程 > 原文： [https://howtodoinjava.com/spring-webflux/spring-webflux-tutorial/](https://howtodoinjava.com/spring-webflux/spring-webflux-tutorial/) Spring 5.0 添加了反應堆 Web 框架 **Spring WebFlux** 。它是完全無阻塞的，支持[響應式](http://www.reactive-streams.org/)背壓，并在 Netty，Undertow 和 Servlet 3.1+ 容器等服務器上運行。在這個 **spring webflux 教程**中，我們將學習響應式編程，webflux api 和功能齊全的 hello world 示例背后的基本概念。 ## 1\. 響應式編程響應式編程是一種編程示例，可促進異步，非阻塞，事件驅動的數據處理方法。響應式編程涉及將數據和事件建模為可觀察的數據流，并實現數據處理例程以對這些流中的更改做出反應。在深入研究響應式世界之前，首先要了解阻塞與非阻塞請求處理之間的區別。 #### 1.1. 阻塞與非阻塞（異步）請求處理 ###### 1.1.1. 阻塞請求處理在傳統的 MVC 應用程序中，當請求到達服務器時，將創建一個 servlet 線程。它將請求委托給工作線程進行 I/O 操作（例如數據庫訪問等）。在工作線程忙時，servlet 線程（請求線程）保持等待狀態，因此被阻塞。也稱為**同步請求處理**。 ![Blocking request processing](https://img.kancloud.cn/d7/f7/d7f721fb2147ad2a818eba5d043612bd_1236x412.jpg) 阻塞請求處理由于服務器可以有一定數量的請求線程，因此限制了服務器在最大服務器負載下處理該數量請求的能力。它可能會影響性能并限制服務器功能的完全利用。 ###### 1.1.2. 非阻塞請求處理在非阻塞或異步請求處理中，沒有線程處于等待狀態。通常只有一個請求線程接收該請求。所有傳入的請求都帶有事件處理器和回調信息。請求線程將傳入的請求委托給線程池（通常為少量線程），該線程池將請求委托給其處理函數，并立即開始處理來自請求線程的其他傳入請求。處理器功能完成后，池中的線程之一將收集響應并將其傳遞給回調函數。 ![Non-blocking request processing](https://img.kancloud.cn/12/8f/128fdf16661cee742d106f8793dee292_1198x438.jpg) 非阻塞請求處理線程的非阻塞性質有助于擴展應用程序的性能。線程數量少意味著內存利用率較低，上下文切換也較少。 #### 1.2. 什么是響應式編程？術語“響應式”是指圍繞響應變化而構建的編程模型。它是基于發布者-訂閱者模式（[觀察者模式](https://howtodoinjava.com/design-patterns/behavioral/observer-design-pattern/)）構建的。在響應式編程中，我們請求資源并開始執行其他事情。當數據可用時，我們會收到通知以及回調函數的數據通知。在回調函數中，我們根據應用程序/用戶需求處理響應。要記住的重要一件事是背壓。在非阻塞代碼中，**控制事件的速率**變得很重要，這樣快速生成器就不會淹沒其目的地。響應式 Web 編程非常適合具有流數據的應用程序以及使用該數據并將其流傳輸給用戶的客戶端。這對開發傳統的 CRUD 應用程序不是很好。如果您要開發具有大量數據的下一個 Facebook 或 Twitter ，那么響應式 API 可能正是您想要的。 ## 2\. 響應式流 API 新的響應式流式 API 由 Netflix，Pivotal，Lightbend，RedHat，Twitter 和 Oracle 等工程師創建，現已成為 Java 9 的一部分。它定義了四個接口： * [`Publisher`](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/api/src/main/java/org/reactivestreams/Publisher.java): 根據從訂閱者收到的需求向訂閱者發出一系列事件。一個發布者可以為多個訂閱者提供服務。它只有一個方法： `Publisher.java` ```java public interface Publisher<T> { public void subscribe(Subscriber<? super T> s); } ``` * [`Subscriber`](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/api/src/main/java/org/reactivestreams/Subscriber.java): 接收和處理發布者發出的事件。請注意，在調用`Subscription#request(long)`發出信號表示需求之前，不會收到任何通知。它有四種方法來處理收到的各種響應。 `Subscriber.java` ```java public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); } ``` * [`Subscription`](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/api/src/main/java/org/reactivestreams/Subscription.java): 定義“發布者”和“訂閱者”之間的一對一關系。只能由單個“訂閱者”使用一次。它既用于表示對數據的需求，又用于取消需求（并允許清除資源）。 `Subscription.java` ```java public interface Subscription<T> { public void request(long n); public void cancel(); } ``` * [`Processor`](https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.2/api/src/main/java/org/reactivestreams/Processor.java): 表示由“訂閱者”和“發布者”組成的處理階段，并服從兩者的契約。 `Processor.java` ```java public interface Processor<T, R> extends Subscriber<T>, Publisher<R> { } ``` > 響應式的兩種流行實現是 [**RxJava**](https://howtodoinjava.com/rxjava/rxjava-tutorial/)（<https://github.com/ReactiveX/RxJava>）和 **Project Reactor**（<https://projectreactor.io/>）。 ## 3\. 什么是 Spring WebFlux？ Spring WebFlux 是 [Spring MVC](https://howtodoinjava.com/spring-mvc-tutorial/) 的并行版本，并支持完全無阻塞的響應式。它支持背壓概念，并使用 [**Netty**](https://netty.io/) 作為內置服務器來運行響應式應用程序。如果您熟悉 Spring MVC 編程風格，則也可以輕松地使用 webflux。 Spring Webflux 使用項目反應器作為反應庫。 Reactor 是 Reactive Streams 庫，因此，它的所有運算符都支持無阻塞背壓。它是與 Spring 緊密合作開發的。 Spring WebFlux 大量使用兩個發布者： * `Mono`：返回 0 或 1 個元素。 ```java Mono<String> mono = Mono.just("Alex"); Mono<String> mono = Mono.empty(); ``` * `Flux`：返回`0…N`個元素。 `Flux`可以是無窮無盡的，這意味著它可以永遠保持發光。它還可以返回一系列元素，然后在返回所有元素后發送完成通知。 ```java Flux<String> flux = Flux.just("A", "B", "C"); Flux<String> flux = Flux.fromArray(new String[]{"A", "B", "C"}); Flux<String> flux = Flux.fromIterable(Arrays.asList("A", "B", "C")); //To subscribe call method flux.subscribe(); ``` 在 Spring WebFlux 中，我們稱為響應式 API /函數，它們返回 mono 和 flux，而您的控制器將返回 mono 和 flux。當您調用返回單聲道或流量的 API 時，它將立即返回。當它們可用時，函數調用的結果將通過單聲道或磁通傳遞給您。 > 要構建真正的非阻塞應用程序，我們必須致力于將其所有組件創建/使用為非阻塞程序，即客戶端，控制器，中間服務甚至數據庫。如果其中之一阻止了請求，我們的目標將會失敗。 ## 4\. Spring Boot WebFlux 示例在此 [**Spring Boot 2**](https://howtodoinjava.com/spring-boot-tutorials/) 應用程序中，我正在創建員工管理系統。我選擇它是因為在學習時，您可以將其與傳統的 MVC 風格的應用程序進行比較。為了使其完全無阻塞，我使用 **mongodb** 作為后端數據庫。 #### 4.1. Maven 依賴包括`spring-boot-starter-webflux`，`spring-boot-starter-data-mongodb-reactive`，`spring-boot-starter-test`和`reactor-test`依賴項。 `pom.xml` ```java <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> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>2.1.1.RELEASE</version> <relativePath />  </parent> <groupId>com.howtodoinjava</groupId> <artifactId>spring-webflux-demo</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>spring-webflux-demo</name> <url>http://maven.apache.org</url> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-webflux</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>io.projectreactor</groupId> <artifactId>reactor-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>javax.xml.bind</groupId> <artifactId>jaxb-api</artifactId> <version>2.3.0</version> </dependency> <dependency> <groupId>javax.servlet</groupId> <artifactId>javax.servlet-api</artifactId> <version>3.1.0</version> <scope>provided</scope> </dependency> </dependencies> </project> ``` #### 4.2. 配置 **Webflux 配置** `WebFluxConfig.java` ```java import org.springframework.context.annotation.Configuration; @Configuration @EnableWebFlux public class WebFluxConfig implements WebFluxConfigurer { } ``` **MongoDb 配置** `MongoConfig.java` ```java import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.data.mongodb.config.AbstractReactiveMongoConfiguration; import org.springframework.data.mongodb.core.ReactiveMongoTemplate; import org.springframework.data.mongodb.repository.config.EnableReactiveMongoRepositories; import com.mongodb.reactivestreams.client.MongoClient; import com.mongodb.reactivestreams.client.MongoClients; @Configuration @EnableReactiveMongoRepositories(basePackages = "com.howtodoinjava.demo.dao") public class MongoConfig extends AbstractReactiveMongoConfiguration { @Value("${port}") private String port; @Value("${dbname}") private String dbName; @Override public MongoClient reactiveMongoClient() { return MongoClients.create(); } @Override protected String getDatabaseName() { return dbName; } @Bean public ReactiveMongoTemplate reactiveMongoTemplate() { return new ReactiveMongoTemplate(reactiveMongoClient(), getDatabaseName()); } } ``` **應用程序配置** `AppConfig.java` ```java import org.springframework.beans.factory.config.PropertyPlaceholderConfigurer; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.core.io.ClassPathResource; @Configuration public class AppConfig { @Bean public static PropertyPlaceholderConfigurer getPropertyPlaceholderConfigurer() { PropertyPlaceholderConfigurer ppc = new PropertyPlaceholderConfigurer(); ppc.setLocation(new ClassPathResource("application.properties")); ppc.setIgnoreUnresolvablePlaceholders(true); return ppc; } } ``` **屬性文件** `application.properties` ```java port=27017 dbname=testdb ``` **日志配置** `logback.xml` ```java <configuration> <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender"> <encoder> <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{5} - %msg%n </pattern> </encoder> </appender> <logger name="org.springframework" level="DEBUG" additivity="false"> <appender-ref ref="STDOUT" /> </logger> <root level="ERROR"> <appender-ref ref="STDOUT" /> </root> </configuration> ``` **Spring Boot 應用程序** `WebfluxFunctionalApp.java` ```java import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class WebfluxFunctionalApp { public static void main(String[] args) { SpringApplication.run(WebfluxFunctionalApp.class, args); } } ``` #### 4.3. REST 控制器 `EmployeeController.java` ```java import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.http.MediaType; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestMethod; import org.springframework.web.bind.annotation.ResponseBody; import org.springframework.web.bind.annotation.ResponseStatus; import org.springframework.web.bind.annotation.RestController; import com.howtodoinjava.demo.model.Employee; import com.howtodoinjava.demo.service.EmployeeService; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @RestController public class EmployeeController { @Autowired private EmployeeService employeeService; @RequestMapping(value = { "/create", "/" }, method = RequestMethod.POST) @ResponseStatus(HttpStatus.CREATED) @ResponseBody public void create(@RequestBody Employee e) { employeeService.create(e); } @RequestMapping(value = "/{id}", method = RequestMethod.GET) @ResponseBody public ResponseEntity<Mono<Employee>> findById(@PathVariable("id") Integer id) { Mono<Employee> e = employeeService.findById(id); HttpStatus status = e != null ? HttpStatus.OK : HttpStatus.NOT_FOUND; return new ResponseEntity<Mono<Employee>>(e, status); } @RequestMapping(value = "/name/{name}", method = RequestMethod.GET) @ResponseBody public Flux<Employee> findByName(@PathVariable("name") String name) { return employeeService.findByName(name); } @RequestMapping(method = RequestMethod.GET, produces = MediaType.TEXT_EVENT_STREAM_VALUE) @ResponseBody public Flux<Employee> findAll() { Flux<Employee> emps = employeeService.findAll(); return emps; } @RequestMapping(value = "/update", method = RequestMethod.PUT) @ResponseStatus(HttpStatus.OK) public Mono<Employee> update(@RequestBody Employee e) { return employeeService.update(e); } @RequestMapping(value = "/delete/{id}", method = RequestMethod.DELETE) @ResponseStatus(HttpStatus.OK) public void delete(@PathVariable("id") Integer id) { employeeService.delete(id).subscribe(); } } ``` #### 4.4. 服務類 `IEmployeeService.java` ```java import com.howtodoinjava.demo.model.Employee; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; public interface IEmployeeService { void create(Employee e); Mono<Employee> findById(Integer id); Flux<Employee> findByName(String name); Flux<Employee> findAll(); Mono<Employee> update(Employee e); Mono<Void> delete(Integer id); } ``` `EmployeeService.java` ```java import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import com.howtodoinjava.demo.dao.EmployeeRepository; import com.howtodoinjava.demo.model.Employee; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; @Service public class EmployeeService implements IEmployeeService { @Autowired EmployeeRepository employeeRepo; public void create(Employee e) { employeeRepo.save(e).subscribe(); } public Mono<Employee> findById(Integer id) { return employeeRepo.findById(id); } public Flux<Employee> findByName(String name) { return employeeRepo.findByName(name); } public Flux<Employee> findAll() { return employeeRepo.findAll(); } public Mono<Employee> update(Employee e) { return employeeRepo.save(e); } public Mono<Void> delete(Integer id) { return employeeRepo.deleteById(id); } } ``` #### 4.5. DAO 存儲庫 `EmployeeRepository.java` ```java import org.springframework.data.mongodb.repository.Query; import org.springframework.data.mongodb.repository.ReactiveMongoRepository; import com.howtodoinjava.demo.model.Employee; import reactor.core.publisher.Flux; public interface EmployeeRepository extends ReactiveMongoRepository<Employee, Integer> { @Query("{ 'name': ?0 }") Flux<Employee> findByName(final String name); } ``` #### 4.6. 模型 `Employee.java` ```java import org.springframework.context.annotation.Scope; import org.springframework.context.annotation.ScopedProxyMode; import org.springframework.data.annotation.Id; import org.springframework.data.mongodb.core.mapping.Document; @Scope(scopeName = "request", proxyMode = ScopedProxyMode.TARGET_CLASS) @Document public class Employee { @Id int id; String name; long salary; //Getters and setters @Override public String toString() { return "Employee [id=" + id + ", name=" + name + ", salary=" + salary + "]"; } } ``` ## 5\. 演示啟動應用程序并檢查請求和響應。 * HTTP POST `http://localhost:8080/create` `API 請求 1` ```java { "id":1, "name":"user_1", "salary":101 } ``` `API 請求 2` ```java { "id":2, "name":"user_2", "salary":102 } ``` * HTTP PUT `http://localhost:8080/update` `API 請求` ```java { "id":2, "name":"user_2", "salary":103 } ``` * HTTP GET `http://localhost:8080/` `API 響應` ```java data:{"id":1,"name":"user_1","salary":101} data:{"id":2,"name":"user_2","salary":102} ``` ![Spring WebFlux Demo](https://img.kancloud.cn/d0/22/d022b14f5c4a8e7e1eb5214a2e746c9e_1042x413.jpg) Spring WebFlux 演示請注意，我正在使用 **Postman chrome 瀏覽器擴展**（是阻止客戶端）測試 API。僅當它已經收集了兩個員工的響應時，才會顯示結果。要驗證非阻塞響應功能，請直接在 chrome 瀏覽器中點擊 URL。結果將以事件的形式（**文本/事件流**）一一出現。為了更好地查看結果，請考慮向控制器 API 添加延遲。 ![Spring WebFlux Demo - Event Stream](https://img.kancloud.cn/5f/eb/5feb4cf3d377c9ac02bd140e06f5b313_1013x482.jpg) Spring WebFlux 示例 – 事件流 ## 6\. Spring WebFlux 教程 – 總結 Spring MVC 和 Spring WebFlux 都支持客戶端-服務器體系結構，但是[并發](https://howtodoinjava.com/java-concurrency-tutorial/)模型和用于阻止自然和線程的默認行為存在關鍵差異。在 Spring MVC 中，假定應用程序可以阻止當前線程，而在 webflux 中，默認情況下線程是非阻止的。這是 **spring webflux 與 mvc** 之間的主要區別。反應和非阻塞通常不會使應用程序運行得更快。響應式和非阻塞性的預期好處是能夠以較少的固定數量的線程和較少的內存需求來擴展應用程序。它使應用程序在負載下更具彈性，因為它們以更可預測的方式擴展。請把關于這個 **spring boot webflux 教程**的問題交給我。學習愉快！ [下載源碼](https://howtodoinjava.com/wp-content/downloads/spring-webflux-demo.zip)