【Spring5】使用 Spring Webflux 开发 Reactive 应用

ゞ浴缸里的玫瑰 2022-05-31 07:17 516阅读 0赞

Spring 5 - Spring webflux 是一个新的非堵塞函数式 Reactive Web 框架，可以用来建立异步的，非阻塞，事件驱动的服务，并且扩展性非常好。

把阻塞（不可避免的）风格的代码迁移到函数式的非阻塞 Reactive 风格代码，需要把商业逻辑作为异步函数来调用。这可以参考 Java 8 的方法或者 lambda 表达式。由于线程是非阻塞的，处理能力能被最大化使用。

在发布这篇文章的时候，Spring 5 还处于一个里程碑版本中（5.0.0 M5）。

## 创建一个Spring Boost项目 ##

可以通过 Spring initializer 创建一个Spring Boot项目。将如下的依赖添加到 pom.xml 中

**\[html\]**  [view plain][]   [copy][view plain]

1.  **<****dependencies****>**  
2.      **<****dependency****>**  
3.          **<****groupId****>**org.springframework.boot**</****groupId****>**  
4.          **<****artifactId****>**spring-boot-starter**</****artifactId****>**  
5.      **</****dependency****>**  
6.      **<****dependency****>**  
7.          **<****groupId****>**org.springframework.boot**</****groupId****>**  
8.          **<****artifactId****>**spring-boot-starter-webflux**</****artifactId****>**  
9.      **</****dependency****>**  
10.     **<****dependency****>**  
11.         **<****groupId****>**org.springframework.boot**</****groupId****>**  
12.         **<****artifactId****>**spring-boot-starter-test**</****artifactId****>**  
13.         **<****scope****>**test**</****scope****>**  
14.     **</****dependency****>**  
15. **</****dependencies****>**

Spring-boot-starter-webflux 包中带了 spring-webflux, netty。其他的依赖需要自行添加。

建立一个简单的用户数据表和从 list 中获取到 user 数据的 DTO 类。这仅仅是一个虚拟的数据 bean，但是这可以实时从其它的数据源像 Rdbms，MongoDb，或者 RestClient 加载数据。由于 JDBC 天生不是响应式的，所以任何对数据库的调用都会阻塞这个线程。MongoDB 有一个响应式的客户端驱动。在测试响应式 Web 服务时的进一步渲染时，REST 风格的调用不会导致任何的阻塞。

**\[java\]**  [view plain][]   [copy][view plain]

1.  **public** **class** User \{  
2.  **public** User()\{\}  
3.    
4.  **public** User(Long id, String user) \{  
5.  **this**.id = id;  
6.  **this**.user = user;  
7.  \}  
8.    
9.  **private** Long id;  
10. **private** String user;  
11.   
12. **public** Long getId() \{ **return** id; \}  
13. **public** **void** setId(Long id) \{ **this**.id = id; \}  
14. **public** String getUser() \{ **return** user; \}  
15. **public** **void** setUser(String user) \{ **this**.user = user; \}  
16. \}  
17.   
18. @Repository  
19. **public** **class** UserRepository \{  
20. **private** **final** List<User> users = Arrays.asList(**new** User(1L, "User1"), **new** User(2L, "User2"));  
21.   
22. **public** Mono<User> getUserById(String id) \{  
23. **return** Mono.justOrEmpty(users.stream().filter(user -> \{  
24. **return** user.getId().equals(Long.valueOf(id));  
25. \}).findFirst().orElse(**null**));  
26. \}  
27.   
28. **public** Flux<User> getUsers() \{  
29. **return** Flux.fromIterable(users);  
30. \}  
31. \}

Mono 和 Flux 是由目标反应器提供的响应类型。Springs 还提供其他的响应流的实现，例如 RXJava。  
  
Mono 和 Flux 是 Reactive streams 的发布者实现。Mono 是 0 或者任意单个值的发布，Flux 是 0 到任意值的发布。他们和 RXJava 中的 Flowable 和 Observable 类似。他们代替流向这些订阅者发布信息。

GetUserById() 返回一个 Mono<User> 对象，这个对象不论何时都会返回 0~1 个用户对象，GetUsers() 返回一连串变动的用户对象，不论何时都包含 0~n 个用户对象。

相比命令式编程风格，我们并不返回可用前阻塞线程的 User/List<User> 对象，而只是返回一个流的引用，流可以在后面访问 User/List<User>。

## 创建带有处理 HTTP 请求函数的 Handler 类 ##

**\[java\]**  [view plain][]   [copy][view plain]

1.  @Service  
2.  **public** **class** UserHandler \{  
3.  @Autowired  
4.  **private** UserRepository userRepository;  
5.    
6.  **public** Mono<ServerResponse> handleGetUsers(ServerRequest request) \{  
7.  **return** ServerResponse.ok().body(userRepository.getUsers(), User.**class**);  
8.  \}  
9.    
10. **public** Mono<ServerResponse> handleGetUserById(ServerRequest request) \{  
11. **return** userRepository.getUserById(request.pathVariable("id"))  
12. .flatMap(user -> ServerResponse.ok().body(Mono.just(user), User.**class**))  
13. .switchIfEmpty(ServerResponse.notFound().build());  
14. \}  
15. \}

handler 类就像 Spring Web 中的 Service beans 一样，我们需要编写该服务的大部分业务功能。ServerResponse 就像 Spring Web 中的 ResponseEntity 类一样，我们可以在 ServerResponse 对象中打包 Response 的数据、状态码、头信息等。 ServerResponse 有很多有用的默认方法，如  `notFound()` ,  `ok()` ,  `accepted()` ,  `created()` 等，可用于创建不同类型的反馈。  
UserHandler 有不同的方法，都返回 Mono<ServerResponse>; UserRepository.getUsers() 返回Flux<User>; 和  `ServerResponse.ok().body(UserRepository.getUsers(), User.class)`  可将此 Flux <User> 转换为 Mono<ServerResponse>，这表明只要可用时均可发起 ServerResponse 的流。 `UserRepository.getUserById()` 返回一个Mono<User>， `ServerResponse.ok().body(Mono.just(user), User.class)`  将此 Mono<User> 转换为Mono<ServerResponse>，这说明随时都可以发起 ServerResponse 的流。

在给定的路径变量（pathVariable）中没有找到用户时，ServerResponse.notFound().build() 返回一个 Mono<ServerResponse>，表名是一个返回 404 服务响应的流。

在命令式编程风格中，数据接收前线程会一直阻塞，这样使得其线程在数据到来前无法运行。而响应式编程中，我们定义一个获取数据的流，然后定义一个在数据到来后的回调函数操作。这样不会使线程堵塞，在数据被返回时，可用线程就用于执行。

## 创建一个定义应用程序路由的路由类 ##

**\[java\]**  [view plain][]   [copy][view plain]

1.  **import** **static** org.springframework.web.reactive.function.server.RequestPredicates.GET;  
2.  **import** **static** org.springframework.web.reactive.function.server.RequestPredicates.accept;  
3.  **import** **static** org.springframework.web.reactive.function.server.RouterFunctions.route;  
4.  @Configuration  
5.  **public** **class** Routes \{  
6.  **private** UserHandler userHandler;  
7.    
8.      **public** Routes(UserHandler userHandler) \{  
9.  **this**.userHandler = userHandler;  
10. \}  
11.   
12. @Bean  
13. **public** RouterFunction<?> routerFunction() \{  
14. **return** route(GET("/api/user").and(accept(MediaType.APPLICATION\_JSON)), userHandler::handleGetUsers)  
15. .and(route(GET("/api/user/\{id\}").and(accept(MediaType.APPLICATION\_JSON)), userHandler::handleGetUserById));  
16. \}  
17. \}

RouterFunction就像Spring Web中的@RequestMapping类一样。 RouterFunction用于定义Spring5应用程序的路由。 RouterFunctions帮助器类有一个有用的方法，类似路由，可用于定义路由并构建RouterFunction对象。 RequestPredicates有许多有用的方法，如GET，POST，path，queryParam，accept，headers，contentType等，来定义路由并构建RouterFunction。 每个路由映射到一个处理程序方法，当接收到适当的HttpRequest时，该方法必须被调用。  
Spring5还支持定义应用程序处理程序映射的@RequestMapping类型的控制器。 我们可以编写如下所示的控制器方法，以在@RequestMapping样式中创建类似的API。

**\[java\]**  [view plain][]   [copy][view plain]

1.  @GetMapping("/user") **public** Mono<ServerResponse> handleGetUsers() \{\}

控制器方法返回Mono<ServerResponse>。  
  
RouterFunction为应用程序提供了DSL类型的路由功能。 到目前为止，Springs不支持混合这两种类型。

## 创建HttpServerConfig类，用于创建HttpServer类 ##

**\[java\]**  [view plain][]   [copy][view plain]

1.  **import** org.springframework.http.server.reactive.HttpHandler;  
2.  **import** org.springframework.http.server.reactive.ReactorHttpHandlerAdapter;  
3.  **import** reactor.ipc.netty.http.server.HttpServer;  
4.  @Configuration  
5.  **public** **class** HttpServerConfig \{  
6.  @Autowired  
7.  **private** Environment environment;  
8.    
9.  @Bean  
10. **public** HttpServer httpServer(RouterFunction<?> routerFunction) \{  
11. HttpHandler httpHandler = RouterFunctions.toHttpHandler(routerFunction);  
12. ReactorHttpHandlerAdapter adapter = **new** ReactorHttpHandlerAdapter(httpHandler);  
13. HttpServer server = HttpServer.create("localhost", Integer.valueOf(environment.getProperty("server.port")));  
14. server.newHandler(adapter);  
15. **return** server;  
16. \}  
17. \}

这将使用应用程序属性中定义的端口创建一个 netty HttpServer。Spring 支持的其他服务器也跟 Tomcat 和 undertow 一样。由于 netty 是异步的，而且天生基于事件驱动，因此更适合响应式的应用程序。Tomcat 使用 Java NIO 来实现 servlet 规范。Netty 是 NIO 的一个实现，它针对异步、事件驱动的非阻塞 IO 应用程序进行了优化。

Tomcat 服务器也可以按照如下代码所示的用法使用：

**\[java\]**  [view plain][]   [copy][view plain]

1.  Tomcat tomcatServer = **new** Tomcat();  
2.      tomcatServer.setHostname("localhost");  
3.      tomcatServer.setPort(Integer.valueOf(environment.getProperty("server.port")));  
4.      Context rootContext = tomcatServer.addContext("", System.getProperty("java.io.tmpdir"));  
5.      ServletHttpHandlerAdapter servlet = **new** ServletHttpHandlerAdapter(httpHandler);  
6.      Tomcat.addServlet(rootContext, "httpHandlerServlet", servlet);  
7.      rootContext.addServletMapping("/", "httpHandlerServlet");  
8.      tomcatServer.start();

## 创建用于启动应用的Spring启动主类 ##

**\[java\]**  [view plain][]   [copy][view plain]

1.  @SpringBootApplication  
2.  **public** **class** Spring5ReactiveApplication \{  
3.  **public** **static** **void** main(String\[\] args) **throws** IOException \{  
4.  SpringApplication.run(Spring5ReactiveApplication.**class**, args);  
5.  \}  
6.  \}

## 测试应用 ##

你可以使用任意诸如Postman、CURL等的HTTP测试工具测试该应用。

Spring测试也支持为响应式服务编写集成测试的功能。

**\[java\]**  [view plain][]   [copy][view plain]

1.  @RunWith(SpringRunner.**class**)  
2.  @SpringBootTest(webEnvironment = WebEnvironment.DEFINED\_PORT)  
3.  **public** **class** UserTest \{  
4.  @Autowired  
5.  **private** WebTestClient webTestClient;  
6.    
7.  @Test  
8.  **public** **void** test() **throws** IOException \{  
9.  FluxExchangeResult<User> result = webTestClient.get().uri("/api/user").accept(MediaType.APPLICATION\_JSON)  
10. .exchange().returnResult(User.**class**);  
11. **assert** result.getStatus().value() == 200;  
12. List<User> users = result.getResponseBody().collectList().block();  
13. **assert** users.size() == 2;  
14. **assert** users.iterator().next().getUser().equals("User1");  
15. \}  
16.   
17. @Test  
18. **public** **void** test1() **throws** IOException \{  
19. User user = webTestClient.get().uri("/api/user/1")  
20. .accept(MediaType.APPLICATION\_JSON).exchange().returnResult(User.**class**).getResponseBody().blockFirst();  
21. **assert** user.getId() == 1;  
22. **assert** user.getUser().equals("User1");  
23. \}  
24.   
25. @Test  
26. **public** **void** test2() **throws** IOException \{  
27. webTestClient.get().uri("/api/user/10").accept(MediaType.APPLICATION\_JSON).exchange().expectStatus()  
28. .isNotFound();  
29. \}  
30. \}

WebTestClient 和 TestRestTemplate 类似， 他们都有调用 Spring 启动应用的 rest 方法，并能够验证响应结果。在 test 的配置中，Spring 测试创建了一个 TestRestTemplate 的 bean。这里面有一个 WebClient，就跟 Spring Web 中的 RestTemplate 类似。这可用于处理响应式和非阻塞的 rest 调用。

**\[java\]**  [view plain][]   [copy][view plain]

1.  WebClient.create("http://localhost:9000").get().uri("/api/user/1")  
2.          .accept(MediaType.APPLICATION\_JSON).exchange().flatMap(resp -> resp.bodyToMono(User.**class**)).block();

exchange()返回Mono<ClientResponse>，它在Emits clientResponse可用时表示一个流。

block()阻塞线程执行，直到Mono返回User/List<User>，因为这是我们需要数据来验证响应的测试用例。  
  
Spring Web 因其易于开发/调试而是必要的。使用Spring5响应式或Spring Web命令式服务的决定必须根据用例明智地做出。在许多情况下，只有命令式的可能会很好，但是在高可扩展性是关键因素时，响应式非阻塞将更适合。

[view plain]: http://blog.csdn.net/hj7jay/article/details/73200882#