Spring Boot 2.0-WebFlux framework
1、介绍
1.1 什么是响应式编程（Reactive Programming）？
简单来说，响应式编程是针对异步和事件驱动的非阻塞应用程序，并且需要少量线程来垂直缩放（即在 JVM 内）而不是水平（即通过集群）。
响应式应用的一个关键方面是“背压（backpressure）”的概念，这是确保生产者不会压倒消费者的机制。例如，当HTTP连接太慢时，从数据库延伸到HTTP响应的反应组件的流水线、数据存储库也可以减慢或停止，直到网络容量释放。
响应式编程也导致从命令式到声明异步组合逻辑的重大转变。与使用Java 8的
CompletableFuture
编写封锁代码相比，可以通过 lambda 表达式编写后续操作。
1.2 响应式 API（Reactive API）和 构建块（Building Blocks）
Spring Framework 5 将 Reactive Streams 作为通过异步组件和库进行背压通信的合同。Reactive Streams 是通过行业协作创建的规范，也已在Java 9中被采用为
。
Spring Framework 在内部使用 Reactor 自己的响应支持。Reactor 是一个 Reactive Streams 实现，进一步扩展基本的 Reactive Streams Publisher 、Flux 和 Mono 可组合的API类型，以提供对
0..N
和
0..1
的数据序列的声明性操作。
Spring Framework 在许多自己的 Reactive API 中暴露了 Flux 和 Mono。然而，在应用级别，一如既往，Spring 提供了选择，并完全支持使用RxJava。有关的更多信息，请查看 Sebastien Deleuze 发表的 "Understanding Reactive Types" 。
2、Spring WebFlux 模块
Spring Framework 5 包括一个新的 spring-webflux 模块。该模块包含对响应式 HTTP 和 WebSocket 客户端的支持，以及对REST，HTML浏览器和 WebSocket风格交互的响应式服务器Web应用程序的支持。
2.1、服务器端
在服务器端 WebFlux 支持2种不同的编程模型：
基于注解的
@Controller
和其他注解也支持 Spring MVC
Functional 、Java 8 lambda 风格的路由和处理
WebFlux 可以在支持 Servlet 3.1 非阻塞 IO API 以及其他异步运行时（如 Netty 和 Undertow ）的 Servlet 容器上运行。每个运行时都适用于响应型
ServerHttpRequest
和
ServerHttpResponse
，将请求和响应的正文暴露为
Flux
，而不是具有响应背压的
InputStream
和
OutputStream
。顶部作为
Flux
支持REST风格的 JSON 和 XML 序列化和反序列化，HTML视图呈现和服务器发送事件也是如此。
基于注解的编程模式
WebFlux中也支持相同的
@Controller
编程模型和 Spring MVC 中使用的相同注解。主要区别在于底层核心框架契约（即
HandlerMappingHandlerAdapter
）是非阻塞的，并且在响应型
ServerHttpRequest
和
ServerHttpResponse
上运行，而不是在
HttpServletRequest
和
HttpServletResponse
上运行。以下是一个响应式 Controller 的例子：
@RestController
public class PersonController {
private final PersonRepository repository;
public PersonController(PersonRepository repository) {
this.repository = repository;
}
@PostMapping("/person")
Mono create(@RequestBody Publisher
personStream) {
}
@GetMapping("/person")
Flux
list() {
}
@GetMapping("/person/")
Mono
findById(@PathVariable String id) {
}
}
函数式编程模式
HandlerFunctions
传入的HTTP请求由
HandlerFunction
处理，
HandlerFunction
本质上是一个接收
ServerRequest
并返回
Mono
的函数。处理函数的注解对应方法将是一个
@RequestMapping
的方法。
ServerRequest
和
ServerResponse
是提供JDK-8友好访问底层HTTP消息的不可变接口。两者都通过在反应堆顶部建立完全反应：请求将身体暴露为
Flux
或
Mono
; 响应接受任何
ReactiveStreamsPublisher
作为主体。
ServerRequest
可以访问各种HTTP请求元素：方法，URI，查询参数，以及通过单独的
ServerRequest.Headers
接口 - 头。通过 body方法 提供对 body 的访问。例如，这是如何将请求体提取为
Mono
：
Mono string = request.bodyToMono(String.class);
这里是如何将身体提取为
Flux
，其中
Person
是可以从body内容反序列化的类（即如果body包含JSON，则由Jackson支持，或者如果是XML，则为JAXB）。
Flux
people = request.bodyToFlux(Person.class);
上面的两个方法（
bodyToMono
和
bodyToFlux
）实际上是使用通用
ServerRequest.body
（
BodyExtractor
）函数的便利方法。
BodyExtractor
是一个功能策略界面，允许您编写自己的提取逻辑，但在
BodyExtractors
实用程序类中可以找到常见的
BodyExtractor
实例。所以，上面的例子可以替换为：
Mono string = request.body(BodyExtractors.toMono(String.class);
Flux
people = request.body(BodyExtractors.toFlux(Person.class);
类似地，
ServerResponse
提供对HTTP响应的访问。由于它是不可变的，您可以使用构建器创建一个
ServerResponse
。构建器允许您设置响应状态，添加响应标题并提供正文。例如，这是如何使用200 OK状态创建响应，JSON内容类型和正文：
Mono
person = ...
ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(person);
这里是如何使用201创建的状态，位置标题和空白体来构建响应：
URI location = ...
ServerResponse.created(location).build();
将这些组合在一起可以创建一个
HandlerFunction
。例如，这里是一个简单的“Hello World”处理程序 lambda 的示例，它返回一个200状态的响应和一个基于 String 的主体：
HandlerFunction helloWorld =
request -> ServerResponse.ok().body(fromObject("Hello World"));
使用 lambda 写处理函数，就像我们上面所说的那样很方便，但是在处理多个函数时可能缺乏可读性，变得不那么容易维护。因此，建议将相关处理函数分组到一个处理程序或控制器类中。例如，这是一个暴露了一个响应式的
Person
存储库的类：
import static org.springframework.http.MediaType.APPLICATION_JSON;
public class PersonHandler {
private final PersonRepository repository;
public PersonHandler(PersonRepository repository) {
this.repository = repository;
}
// 1
public Mono listPeople(ServerRequest request) {
Flux
people = repository.allPeople();
return ServerResponse.ok().contentType(APPLICATION_JSON).body(people, Person.class);
}
// 2
public Mono createPerson(ServerRequest request) {
Mono
person = request.bodyToMono(Person.class);
return ServerResponse.ok().build(repository.savePerson(person));
}
// 3
public Mono getPerson(ServerRequest request) {
int personId = Integer.valueOf(request.pathVariable("id"));
Mono notFound = ServerResponse.notFound().build();
return personMono
.flatMap(person -> ServerResponse.ok().contentType(APPLICATION_JSON).body(fromObject(person)))
.switchIfEmpty(notFound);
}
}
1/
listPeople
是一个处理函数，它将数据库中发现的所有
Person
对象返回为JSON。
2/
createPerson
是一个处理函数，用于存储请求正文中包含的新
Person
。请注意，
PersonRepository.savePerson(Person)
返回
Mono
：发出完成信号的空
Mono
，当人从请求中读取并存储时，发出完成信号。因此，当接收到完成信号时，即当
Person
已被保存时，我们使用
build(Publisher)
方法来发送响应。
3/
getPerson
是一个处理函数，它通过路径变量id来标识一个人。我们通过数据库检索该
Person
，并创建一个JSON响应（如果找到）。如果没有找到，我们使用
switchIfEmpty(Mono)
来返回 404 Not Found 响应。
RouterFunctions
传入请求将路由到处理函数，并使用一个
RouterFunction
，它是一个服务器
ServerRequest
的函数，并返回一个
Mono
。如果请求与特定路由匹配，则返回处理函数; 否则返回一个空的
Mono
。
RouterFunction
与
@Controller
类中的
@RequestMapping
注解类似。
.
通常，您不要自己编写路由器功能，而是使用
RouterFunctions.route(RequestPredicate,HandlerFunction)
， 使用请求谓词和处理函数创建一个。如果谓词适用，请求将路由到给定的处理函数; 否则不执行路由，导致 404 Not Found 响应。虽然您可以编写自己的
RequestPredicate
，但是您不需要：
RequestPredicates
实用程序类提供常用的谓词，基于路径，HTTP方法，内容类型等进行匹配。使用路由，我们可以路由到我们的 “Hello World” 处理函数：
RouterFunction helloWorldRoute =
RouterFunctions.route(RequestPredicates.path("/hello-world"),
request -> Response.ok().body(fromObject("Hello World")));
两个路由功能可以组成一个新的路由功能，路由到任一处理函数：如果第一个路由的谓词不匹配，则第二个被评估。组合的路由器功能按顺序进行评估，因此在通用功能之前放置特定功能是有意义的。您可以通过调用
RouterFunction.and(RouterFunction)
或通过调用
RouterFunction.andRoute(RequestPredicate,HandlerFunction)
来组成两个路由功能，这是
RouterFunction.and()
与
RouterFunctions.route()
的一种方便组合。
给定我们上面显示的
PersonHandler
，我们现在可以定义路由功能，路由到相应的处理函数。我们使用 方法引用（method-references） 来引用处理函数：
import static org.springframework.http.MediaType.APPLICATION_JSON;
PersonRepository repository = ...
PersonHandler handler = new PersonHandler(repository);
RouterFunction personRoute =
route(GET("/person/").and(accept(APPLICATION_JSON)), handler::getPerson)
.andRoute(GET("/person").and(accept(APPLICATION_JSON)), handler::listPeople)
.andRoute(POST("/person").and(contentType(APPLICATION_JSON)), handler::createPerson);
除路由功能之外，您还可以通过调用
RequestPredicate.and(RequestPredicate)
或
RequestPredicate.or(RequestPredicate)
来构成请求谓词。这些工作正如预期的那样：如果给定的谓词匹配，则生成的谓词匹配; 或者如果任一谓词都匹配。
RequestPredicates
中发现的大多数谓词是组合的。例如，
RequestPredicates.GET(String)
是
RequestPredicates.method(HttpMethod)
和
RequestPredicates.path(String)
的组合。
启动服务器
现在只有一个难题遗留：在HTTP服务器中运行路由功能。您可以使用
RouterFunctions.toHttpHandler(RouterFunction)
将路由功能转换为
HttpHandler
。
HttpHandler
允许您运行各种响应场景：Reactor Netty，Servlet 3.1和Undertow。以下是在 Reactor Netty 中运行路由功能的方法，例如：
RouterFunction route = ...
HttpHandler httpHandler = RouterFunctions.toHttpHandler(route);
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(httpHandler);
HttpServer server = HttpServer.create(HOST, PORT);
server.newHandler(adapter).block();
对于 Tomcat ，它看起来像这样：
RouterFunction route = ...
HttpHandler httpHandler = RouterFunctions.toHttpHandler(route);
HttpServlet servlet = new ServletHttpHandlerAdapter(httpHandler);
Tomcat server = new Tomcat();
Tomcat.addServlet(rootContext, "servlet", servlet);
rootContext.addServletMapping("/", "servlet");
tomcatServer.start();
待完成：DispatcherHandler HandlerFilterFunction 路由功能映射的路由可以通过调用
RouterFunction.filter(HandlerFilterFunction)
进行过滤，其中
HandlerFilterFunction
本质上是一个接收
ServerRequest
和
HandlerFunction
的函数，并返回一个
ServerResponse
。处理函数参数表示链中的下一个元素：通常是路由到的
HandlerFunction
，但是如果应用了多个过滤器，也可以是另一个
FilterFunction
。使用注解，可以使用
@ControllerAdvice
和
/
或
ServletFilter
来实现类似的功能。让我们在我们的路由中添加一个简单的安全过滤器，假设我们有一个
SecurityManager
可以确定是否允许特定的路径：
SecurityManager securityManager = ...
RouterFunction route = ...
RouterFunction filteredRoute =
route.filter(request, next) -> {
if (securityManager.allowAccessTo(request.path())) {
return next.handle(request);
}
else {
return ServerResponse.status(UNAUTHORIZED).build();
}
});
在这个例子中可以看到，调用
next.handle(ServerRequest)
是可选的：我们只允许在允许访问时执行处理函数。
2.2 客户端（Client Side）
WebFlux 包括一个 functional, reactive WebClient，它为
RestTemplate
提供了一种完全无阻塞和响应式的替代方案。 它将网络输入和输出公开为客户端
HttpRequest
和
ClientHttpResponse
，其中请求和响应的主体是
Flux
而不是
InputStream
和
OutputStream
。此外，它还支持与服务器端相同的响应式 JSON，XML和SSE 序列化机制，因此您可以使用类型化对象。以下是使用需要
ClientHttpConnector
实现的
WebClient
插入特定HTTP客户端（如 Reactor Netty）的示例：
WebClient client = WebClient.create("http://example.com");
Mono account = client.get()
.url("/accounts/", 1L)
.accept(APPLICATION_JSON)
.exchange(request)
.flatMap(response -> response.bodyToMono(Account.class));
AsyncRestTemplate
还支持非阻塞交互。主要区别在于它不支持非阻塞流，例如 Twitter one ，因为它基本上仍然依赖于
InputStream
和
OutputStream
。
2.4 请求体和响应体的转换（Request and Response Body Conversion）
spring-core
模块提供了响应式
Encoder(编码器)
和
Decoder(解码器)
，使得能够串行化字符串与类型对象的转换。
spring-web
模块添加了 JSON（Jackson）和 XML（JAXB）实现，用于Web应用程序以及其他用于SSE流和零拷贝文件传输。
支持以下 Reactive API：
Reactor 3.x 支持开箱即用
依赖项在类路径上时支持 RxJava 2.x
当 ·io.reactivex:rxjava
和
io.reactivex:rxjava-reactive-streams`（RxJava 和 Reactive Streams 之间的适配器）依赖关系在类路径上时，支持 RxJava 1.x
例如，请求体可以是以下方式之一，它将在注解和功能编程模型中自动解码：
Accountaccount
- 在调用控制器之前，account 将无阻塞地被反序列化。
Monoaccount
- controller 可以使用
Mono
来声明在反序列化 account 后执行的逻辑。
Singleaccount
- 和
Mono
类似，但是用的是 RxJava
Fluxaccounts
- 输入流场景
Observableaccounts
- RxJava 的 输入流场景
响应体（response body）可以是以下之一：
Mono
- 当
Mono
完成时，序列化而不阻塞给定的Account。
Single
- 与上类似，但是使用的 RxJava
Flux
- 流式场景，可能是SSE，具体取决于所请求的内容类型。
Observable
- 与上类似， 但是使用的 RxJava
Observable
类型
Flowable
- 与上类似， 但是使用的 RxJava 2
Flowable
类型。
Publisher
或
Flow.Publisher
- 支持任何实现Reactive Streams
Publisher
的类型。
Flux
- SSE 流。
Mono
- 当
Mono
完成时，请求处理完成。
Account
- 序列化而不阻塞给定的Account; 意味着同步、非阻塞的 Controller 方法。
Void
- 特定于基于注解的编程模型，方法返回时，请求处理完成; 意味着同步、非阻塞的 Controller 方法。
当使用像
Flux
或
Observable
这样的流类型时，请求/响应或映射/路由级别中指定的媒体类型用于确定数据应如何序列化和刷新。例如，返回
Flux
的REST端点将默认序列化如下：
application/json
:
Flux
作为异步集合处理，并在完成事件发布时将其序列化为具有显式刷新的JSON数组。
application/stream+json
: 一个
Flux
将作为一系列的
Account
元素处理，作为以新行分隔的单个JSON对象，并在每个元素之后显式刷新。WebClient 支持JSON流解码，因此这对于服务器到服务器的用例来说是一个很好的用例。
text/event-stream
: 一个
Flux
或
Flux>
将作为一个
Stream
或
ServerSentEvent
元素的流处理，作为单独的 SSE 元素，使用默认的JSON进行数据编码和每个元素之间的显式刷新。这非常适合将流暴露给浏览器客户端。WebClient 也支持读取SSE流。
2.4 响应式 Websocket 支持
WebFlux 包括响应式 WebSocket 客户端和服务器支持。Java WebSocket API（JSR-356），Jetty，Undertow和Reactor Netty都支持客户端和服务器。
在服务器端，声明一个
WebSocketHandlerAdapter
，然后简单地添加映射到基于
WebSocketHandler
的端点：
@Bean
public HandlerMapping webSocketMapping() {
Map map = new HashMap();
map.put("/foo", new FooWebSocketHandler());
map.put("/bar", new BarWebSocketHandler());
SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
mapping.setOrder(10);
mapping.setUrlMap(map);
return mapping;
}
@Bean
public WebSocketHandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter();
}
在客户端，为上面列出的支持的库之一创建一个
WebSocketClient
：
WebSocketClient client = new ReactorNettyWebSocketClient();
client.execute("ws://localhost:8080/echo"), session -> {... }).blockMillis(5000);
2.5 测试
spring-test
模块包括一个
WebTestClient
，可用于测试具有或不具有正在运行的服务器的 WebFlux 服务器端点。
没有运行服务器的测试与来自Spring MVC的
MockMvc
相当，其中使用模拟请求和响应，而不是使用套接字通过网络连接。然而，
WebTestClient
也可以针对正在运行的服务器执行测试。
更多请查看 sample tests
3、开始入门
3.1 Spring Boot Starter
3.2 手动引导（Manual Bootstrapping）
对于依赖关系，从
spring-webflux
和
spring-context
开始。 然后添加
jackson-databind
和
io.netty:netty-buffer
（暂时见SPR-14528）以获得JSON支持。最后添加一个支持的运行时的依赖项：
Tomcat
Jetty
ReactorNetty
Undertow
— io.undertow:undertow-core
基于注解编程模式的引导：
ApplicationContext context = new AnnotationConfigApplicationContext(DelegatingWebFluxConfiguration.class); // (1)
HttpHandler handler = DispatcherHandler.toHttpHandler(context); // (2)
以上加载默认的 Spring Web 框架配置（1），然后创建一个
DispatcherHandler
，主类驱动请求处理（2），并适应
HttpHandler
- 响应式HTTP请求处理的最低级别的Spring抽象。
函数编程模式的引导：
AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext(); // (1)
context.registerBean(FooBean.class, () -> new FooBeanImpl()); // (2)
context.registerBean(BarBean.class); // (3)
context.refresh();
HttpHandler handler = WebHttpHandlerBuilder
.webHandler(RouterFunctions.toHttpHandler(...))
.applicationContext(context)
.build(); // (4)
以上创建了一个
AnnotationConfigApplicationContext
实例（1），可以利用新的功能 bean 注册API（2）使用 Java 8 供应商注册 bean，或者只需通过指定其类（3）即可。
HttpHandler
是使用
WebHttpHandlerBuilder
（4）创建的。
然后可以将
HttpHandler
安装在支持的运行服务器之一中：
// Tomcat and Jetty (also see notes below)
HttpServlet servlet = new ServletHttpHandlerAdapter(handler);
...
// Reactor Netty
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(handler);
HttpServer.create(host, port).newHandler(adapter).block();
// Undertow
UndertowHttpHandlerAdapter adapter = new UndertowHttpHandlerAdapter(handler);
Undertow server = Undertow.builder().addHttpListener(port, host).setHandler(adapter).build();
server.start();
对于特别是使用 WAR 部署的 Servlet 容器，可以使用作为
WebApplicationInitializer
的
AbstractAnnotationConfigDispatcherHandlerInitializer
，并由 Servlet容器自动检测。它负责注册
ServletHttpHandlerAdapter
，如上所示。您将需要实现一个抽象方法来指向您的 Spring 配置。
3.3 Examples
您将在以下项目中找到有助于构建反应式 Web 应用程序的代码示例：
Functional programming model sample
Spring Reactive Playground: playground for most Spring Web reactive features
Reactor website: the spring-functional branch is a Spring 5 functional, Java 8 lambda-style application
Spring Reactive University session: live-coded project from this Devoxx BE 2106 university talk
Reactive Thymeleaf Sandbox
Mix-it 2017 website: Kotlin + Reactive + Functional web and bean registration API application
Reactor by example: code snippets coming from this InfoQ article
Spring integration tests: various features tested with Reactor StepVerifier
本文来自企鹅号 - java交流学习 媒体
如有侵权，请联系 cloudcommunity@tencent.com 删除。
本文来自企鹅号 - java交流学习 媒体
如有侵权，请联系 cloudcommunity@tencent.com 删除。