二、反应式集成-spring

在这里插入图片描述

一、Spring WebFlux

what

1、简介

- Spring WebFlux 包含一个用于执行 HTTP 请求的客户端。 有一个 基于 Reactor 的功能性、流畅的 API，请参阅 Reactive Libraries， 它支持异步逻辑的声明性组合，而无需处理 线程或并发。它是完全无阻塞的，它支持流媒体，并依赖于 也用于编码和 解码服务器端的请求和响应内容。

1、特点

- WebClient需要一个 HTTP 客户端库来执行请求。有内置 支持以下各项：

	- 1、Reactor Netty（内蒂反应堆）

	- 2、JDK HttpClient

	- 3、Jetty 反应式 HttpClient

	- 4、Apache HttpComponents（阿帕奇 HttpComponents）

how

1、配置

1、初始化

1、创建 A 的最简单方法是通过静态工厂方法之一：WebClient

		- 1、WebClient.create()
		- 2、WebClient.create(String baseUrl)

2、还可以使用其他选项：WebClient.builder()

		- uriBuilderFactory：自定义用作基本 URL。UriBuilderFactorydefaultUriVariables：展开 URI 模板时使用的默认值。defaultHeader：每个请求的标头。defaultCookie：每个请求的 Cookie。defaultRequest：自定义每个请求。Consumerfilter：每个请求的客户端过滤器。exchangeStrategies：HTTP 消息读取器/写入器自定义。clientConnector：HTTP 客户端库设置。observationRegistry：用于启用可观测性支持的注册表。observationConvention：一种可选的自定义约定，用于提取记录观测值的元数据。

3、例子：

		- 一旦构建，a 是不可变的。但是，您可以克隆它并构建一个 修改后的副本如下

WebClient client = WebClient.builder()
.codecs(configurer -> … )
.build();
-
WebClient client1 = WebClient.builder()
.filter(filterA).filter(filterB).build();

WebClient client2 = client1.mutate()
.filter(filterC).filter(filterD).build();

// client1 has filterA, filterB

// client2 has filterA, filterB, filterC, filterD

2、配置项：具体配置可看官网

1、MaxInMemorySize

		- 编解码器对缓冲数据有限制 内存以避免应用程序内存问题。默认情况下，这些设置为 256KB。 如果这还不够，您将收到以下错误：

			- org.springframework.core.io.buffer.DataBufferLimitException: Exceeded limit on max bytes to buffer

	- 2、Reactor Netty（内蒂反应堆）
	- 3、资源
	- 4、超时
	- 5、JDK HttpClient
	- 6、Jetty
	- 7、HttpComponents

2、功能：具体可看官网

1、retrieve()

	- 该方法可用于声明如何提取响应

2、Exchange：交换

	- 和 方法（或 Kotlin 中的 and） 对于需要更多控制的更高级情况（例如以不同的方式解码响应）很有用 根据响应状态：exchangeToMono()exchangeToFlux()awaitExchange { }exchangeToFlow { }

3、Request Body：

	- 请求正文可以从 处理的任何异步类型进行编码，由 处理 like 或 Kotlin 协程，如以下示例所示：ReactiveAdapterRegistryMonoDeferred

4、Filters

	- 您可以通过 注册客户端过滤器 （） 以拦截和修改请求，如以下示例所示：ExchangeFilterFunctionWebClient.Builder

5、Attributes

	- 您可以向请求添加属性。如果您想传递信息，这很方便 通过过滤器链并影响给定请求的过滤器的行为。 例如：

6、Context

	- 属性提供了一种将信息传递到筛选器的便捷方式 链，但它们只影响当前请求。如果要传递信息 传播到嵌套的其他请求，例如 via ，或在 之后执行的请求。 例如，via ，那么您需要使用 Reactor 。flatMapconcatMapContext

反应器需要填充在反应链的末端，以便适用于所有操作。例如：Context

7、Synchronous Use

	- WebClient可以通过在末尾阻止结果来同步使用：

		-

Person person = client.get().uri(“/person/{id}”, i).retrieve()
.bodyToMono(Person.class)
.block();

List persons = client.get().uri(“/persons”).retrieve()
.bodyToFlux(Person.class)
.collectList()
.block();

	- 但是，如果需要进行多次调用，则避免对每个调用进行阻塞会更有效 单独响应，而是等待组合结果：

		-

Person person = client.get().uri(“/person/{id}”, i).retrieve()
.bodyToMono(Person.class)
.block();

List persons = client.get().uri(“/persons”).retrieve()
.bodyToFlux(Person.class)
.collectList()
.block();

	- 以上只是一个例子。还有很多其他的模式和运算符用于放置 一起形成一个反应式管道，可以进行许多远程调用，可能有些是嵌套的， 相互依存，直到最后都没有阻塞。

使用或，您永远不必在 Spring MVC 或 Spring WebFlux 控制器中阻塞。只需从控制器方法返回生成的反应类型即可。同样的原则也适用于 Kotlin 协程和 Spring WebFlux，只需在 controller 方法。FluxMonoFlow

8、Testing

	- 若要测试使用 的代码，可以使用模拟 Web 服务器，例如 OkHttp MockWebServer。查看示例 有关其用途，请查看 Spring Framework 测试套件中的 WebClientIntegrationTests 或 OkHttp 存储库中的 static-server 示例。WebClient

二、Spring WebClient

1、what

1、HTTP Interface Client

- 1、简介：Spring Frameworks 允许您将 HTTP 服务定义为具有 HTTP 的 Java 接口 交换方法。然后，您可以生成实现此接口的代理，并 执行交换。这有助于简化 HTTP 远程访问，并提供额外的 灵活选择 API 样式，例如同步式或响应式。

	- 详细信息参考 REST终结点

2、WebSockets

1、简介：

	- WebSocket 协议 RFC 6455 提供了一个标准化的 在客户端和服务器之间建立全双工、双向通信通道的方法 通过单个 TCP 连接。它是一种与 HTTP 不同的 TCP 协议，但旨在 通过 HTTP 工作，使用端口 80 和 443，并允许重用现有防火墙规则。
	- 注意：

		- 请注意，如果 WebSocket 服务器在 Web 服务器（例如 nginx）后面运行，则 可能需要将其配置为将 WebSocket 升级请求传递到 WebSocket 服务器。同样，如果应用程序在云环境中运行，请选中 云提供商与 WebSocket 支持相关的说明。

2、HTTP 与 WebSocket

	- 即使 WebSocket 被设计为与 HTTP 兼容并且以 HTTP 请求开头， 重要的是要了解这两种协议导致非常不同 体系结构和应用程序编程模型。

在 HTTP 和 REST 中，应用程序被建模为多个 URL。要与应用程序交互，客户端访问这些 URL，请求-响应样式。服务器将请求路由基于 HTTP URL、方法和标头的适当处理程序。

相比之下，在 WebSocket 中，初始连接通常只有一个 URL。随后，所有应用程序消息都在同一 TCP 连接上流动。这指向一个完全不同的异步、事件驱动的消息传递体系结构。

WebSocket 也是一种低级传输协议，与 HTTP 不同，它没有规定消息内容的任何语义。这意味着没有办法路由或处理除非客户端和服务器在消息语义上达成一致，否则消息。

WebSocket 客户端和服务器可以协商使用更高级别的消息传递协议（例如，STOMP），通过 HTTP 握手请求上的标头。如果没有这一点，他们需要提出自己的惯例。Sec-WebSocket-Protocol

3、何时使用：

	- WebSocket 可以使网页具有动态性和交互性。但是，在许多情况下， AJAX 和 HTTP 流式处理或长轮询的组合可以提供简单且 有效的解决方案。

例如，新闻、邮件和社交源需要动态更新，但可能需要动态更新每隔几分钟就这样做是完全可以的。协作、游戏和金融应用，在另一方面，需要更接近实时。

延迟本身并不是决定性因素。如果消息量相对较低（例如，监控网络故障）HTTP 流式处理或轮询可以提供有效的解决方案。低延迟、高频率和高音量的结合才能发挥最佳效果使用 WebSocket 的案例。

还要记住，在互联网上，您无法控制的限制性代理可能会阻止 WebSocket 交互，因为它们未配置为传递标头，或者因为它们关闭了看似空闲的长期连接。这意味着将 WebSocket 用于防火墙内的内部应用程序是一个与面向公众的应用程序相比，它更直接。Upgrade

4、特性与用法：

	- Spring Framework 提供了一个 WebSocket API，您可以使用它来编写客户端和 处理 WebSocket 消息的服务器端应用程序。

1、服务器端

		- 若要创建 WebSocket 服务器，可以先创建一个 . 以下示例演示如何执行此操作：WebSocketHandler
		- 然后，您可以将其映射到 URL：

import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;

public class MyWebSocketHandler implements WebSocketHandler {

@Override
public Mono handle(WebSocketSession session) {
// …
}
}
- 如果使用 WebFlux Config，则什么都没有如果不使用 WebFlux 配置，则需要声明如下所示：WebSocketHandlerAdapter
@Configuration
class WebConfig {

@Bean
public HandlerMapping handlerMapping() {
Map<String, WebSocketHandler> map = new HashMap<>();
map.put(“/path”, new MyWebSocketHandler());
int order = -1; // before annotated controllers

return new SimpleUrlHandlerMapping(map, order);
}
}

@Configuration
class WebConfig {

// …

@Bean
public WebSocketHandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter();
}
}

2、WebSocketHandler：具体实现参考官网

1、简介

				- 用于指示会话的应用程序处理何时完成的 take 和 return 方法。会话已处理 通过两个流，一个用于入站消息，一个用于出站消息。下表 介绍处理流的两种方法：handleWebSocketHandlerWebSocketSessionMono

					- 

				- A 必须将入站和出站流组合成一个统一的流，并且 返回反映该流完成的 A。取决于应用 要求，统一流在以下情况下完成：WebSocketHandlerMono

入站或出站消息流完成。入站流完成（即连接关闭），而出站流是无限的。在选定的点上，通过 .closeWebSocketSession
当入站和出站消息流组合在一起时，无需检查连接是否打开，因为 Reactive Streams 会发出结束活动的信号。入站流接收完成或错误信号，出站流接收到完成或错误信号接收取消信号。

2、用法

				- 1、处理程序的最基本实现是处理入站流的实现。这 以下示例演示了这样的实现：

					- 访问入站消息流。对每条消息执行一些操作。执行使用消息内容的嵌套异步操作。返回 a 在接收完成时完成。Mono

class ExampleHandler implements WebSocketHandler {

@Override
public Mono handle(WebSocketSession session) {
return session.receive()
.doOnNext(message -> {
// …
})
.concatMap(message -> {
// …
})
.then();
}
}
- 注意：对于嵌套的异步操作，可能需要调用基础使用池数据缓冲区的服务器（例如，Netty）。否则，数据缓冲区可能是在您有机会阅读数据之前发布。有关更多背景信息，请参阅数据缓冲区和编解码器。message.retain()

				- 2、以下实现将入站流和出站流组合在一起：

					- 处理入站消息流。创建出站消息，生成组合流。返回未完成的 a，而我们继续接收。Mono

class ExampleHandler implements WebSocketHandler {

@Override
public Mono handle(WebSocketSession session) {

Flux output = session.receive()
.doOnNext(message -> {
// …
})
.concatMap(message -> {
// …
})
.map(value -> session.textMessage("Echo " + value));

return session.send(output);
}
}

				- 3、入站和出站流可以是独立的，并且只有在完成时才能加入， 如以下示例所示：

					- 处理入站消息流。发送传出消息。联接流并返回一个在任一流结束时完成的 a。Mono

class ExampleHandler implements WebSocketHandler {

@Override
public Mono handle(WebSocketSession session) {

Mono input = session.receive()
.doOnNext(message -> {
// …
})
.concatMap(message -> {
// …
})
.then();

Flux source = … ;
Mono output = session.send(source.map(session::textMessage));

return Mono.zip(input, output).then();
}
}

3、DataBuffer

		- DataBuffer是 WebFlux 中字节缓冲区的表示形式。Spring Core 部分 参考资料在数据缓冲区和编解码器部分有更多内容。要了解的关键点是，在某些 像 Netty 这样的服务器，字节缓冲区是池化的，引用计数，必须释放 使用时以避免内存泄漏。

在 Netty 上运行时，应用程序必须使用希望保留输入数据缓冲区以确保它们不会被释放，并且随后在缓冲区用完时使用。DataBufferUtils.retain(dataBuffer)DataBufferUtils.release(dataBuffer)

4、握手

		- WebSocketHandlerAdapter委托给 .默认情况下，这是一个实例 的，它对 WebSocket 请求和 然后用于正在使用的服务器。目前，有内置的 支持 Reactor Netty、Tomcat、Jetty 和 Undertow。WebSocketServiceHandshakeWebSocketServiceRequestUpgradeStrategy

HandshakeWebSocketService公开一个属性，该属性允许设置 A 从中提取属性并插入它们转换为 .sessionAttributePredicatePredicateWebSessionWebSocketSession

5、服务器配置

1、简介

			- 对于每个服务器，都公开了特定于 底层 WebSocket 服务器引擎。使用 WebFlux Java 配置时，您可以自定义 WebFlux Config 的相应部分中显示的此类属性，或者如果 不使用 WebFlux 配置，请使用以下命令：RequestUpgradeStrategy

2、用法

			- 检查服务器的升级策略，了解有哪些选项可用。现在 只有 Tomcat 和 Jetty 公开了此类选项。

@Configuration
class WebConfig {

@Bean
public WebSocketHandlerAdapter handlerAdapter() {
return new WebSocketHandlerAdapter(webSocketService());
}

@Bean
public WebSocketService webSocketService() {
TomcatRequestUpgradeStrategy strategy = new TomcatRequestUpgradeStrategy();
strategy.setMaxSessionIdleTimeout(0L);
return new HandshakeWebSocketService(strategy);
}
}

6、 CORS

		- 配置 CORS 并限制对 WebSocket 终结点的访问的最简单方法是 实现并返回包含允许的源、标头和其他详细信息的 A。如果你做不到 那，您还可以将属性设置为 按 URL 模式指定 CORS 设置。如果同时指定了两者，则使用 on 上的方法将它们组合在一起。WebSocketHandlerCorsConfigurationSourceCorsConfigurationcorsConfigurationsSimpleUrlHandlercombineCorsConfiguration

7、客户

1、简介

			- Spring WebFlux 提供了一个抽象，其中包含以下实现 Reactor Netty、Tomcat、Jetty、Undertow 和标准 Java（即 JSR-356）。WebSocketClient
			- note:tomcat 客户端实际上是标准 Java 客户端的扩展，但有一些额外的扩展 处理中的功能，以利用 Tomcat 特定的功能 用于暂停接收背压消息的 API。WebSocketSession

2、用法

			- 若要启动 WebSocket 会话，可以创建客户端实例并使用其方法：execute

				- 某些客户端（如 Jetty）实施并需要停止和启动 在您可以使用它们之前。所有客户端都有与配置相关的构造函数选项 的基础 WebSocket 客户端。Lifecycle

WebSocketClient client = new ReactorNettyWebSocketClient();

URI url = new URI(“ws://localhost:8080/path”);
client.execute(url, session ->
session.receive()
.doOnNext(System.out::println)
.then());

- 5.、由此总结：他们皆都集成反应式

3、RSocket

- RSocket 是一种通过 TCP 进行多路复用双工通信的应用程序协议， WebSocket 和其他字节流传输，使用以下交互之一 模型：

	- Request-Response— 发送一条消息并接收一条消息。Request-Stream— 发送一条消息并接收回一条消息流。Channel— 在两个方向上发送消息流。Fire-and-Forget— 发送单向消息。

- 详细请看官网：

4、反应式库

1、简介

	- spring-webflux依赖于并在内部使用它来编写异步 逻辑，并提供反应式流支持。通常，WebFlux API 返回 or（因为它们在内部使用）并宽松地接受任何 Reactive Streams 实现作为输入。 当提供 a 时，只能将其视为语义未知 （0..N） 的流。 但是，如果语义是已知的，则应将其包装为 or 传递原始. 使用 versus 很重要，因为它有助于表达基数 — 例如，是否需要单个或多个异步值， 这对于做出决策至关重要（例如，在编码或解码 HTTP 消息时）。reactor-coreFluxMonoPublisherPublisherFluxMono.from(Publisher)PublisherFluxMono

2、特性

	- 对于带注释的控制器，WebFlux 透明地适应 应用程序。这是在 ReactiveAdapterRegistry 的帮助下完成的，它 提供对反应式库和其他异步类型的可插拔支持。注册表 内置了对 RxJava 3、Kotlin 协程和 SmallRye Mutiny 的支持，但您可以 也注册其他人。

2、how

1、测试

- 该模块提供 和 的模拟实现。 参见 Spring Web Reactive 论坛 讨论模拟对象。spring-testServerHttpRequestServerHttpResponseServerWebExchange

WebTestClient 基于这些模拟请求和响应对象，为在没有 HTTP 的情况下测试 WebFlux 应用程序提供支持服务器。您也可以将用于端到端集成测试。WebTestClient

XMind - Trial Version