spring全家桶系列之spring boot 2.2-Spring WebFlux框架（四）

小灰灰 2021-12-21 07:39 1082阅读 0赞

### 简介 ###

*  Spring WebFlux是Spring Framework 5.0中引入的新的响应式Web框架。与Spring MVC不同，它不需要Servlet API，完全异步且无阻塞，并 通过Reactor项目实现Reactive Streams规范。基于jdk8的管道流（先学习Lamda表达式）完成异步无阻塞。

### 为什么使用 ###

*  部分答案是需要非阻塞Web堆栈来处理少量线程的并发性，并使用较少的硬件资源进行扩展。Servlet 3.1确实为非阻塞I / O提供了API。但是，使用它会远离Servlet API的其余部分，其中契约是同步（`Filter`，`Servlet`）或阻塞（`getParameter`，`getPart`）。这是新的通用API作为跨任何非阻塞运行时的基础的动机。这很重要，因为在异步，非阻塞空间中已经建立了良好的服务器（如Netty）。
 *  答案的另一部分是函数式编程。就像在Java 5中添加注释一样创造了机会（例如带注释的REST控制器或单元测试），Java 8中添加lambda表达式为Java中的功能API创造了机会。这是非阻塞应用程序和延续风格的API（如由普及的福音`CompletableFuture`和[ReactiveX][]），其允许异步逻辑的声明性组合物。在编程模型级别，Java 8使Spring WebFlux能够提供功能性Web端点以及带注释的控制器。

### **可以替换MVC吗?不可以** ###

*  因为大部分框架设置都是阻塞式的,如果使用spring reactive版本都可以使用例如:spring-boot-starter-data-redis-reactive.
 *  如果你的Spring MVC应用程序运行正常，则无需更改。命令式编程是编写，理解和调试代码的最简单方法。你有最多的库选择，因为从历史上看，大多数库都是阻塞的。
 *  Spring WebFlux提供与此空间中的其他人相同的执行模型优势，并且还提供服务器选择（Netty，Tomcat，Jetty，Undertow和Servlet 3.1+容器），选择编程模型（带注释的控制器和功能Web端点），以及选择反应库（Reactor，RxJava或其他）。
 *  如果你对与Java 8 lambdas或Kotlin一起使用的轻量级，功能性Web框架感兴趣，则可以使用Spring WebFlux功能Web端点。对于较小的应用程序或具有较低复杂要求的微服务而言，这也是一个不错的选择，可以从更高的透明度和控制中受益。
 *  在微服务架构中，你可以将应用程序与Spring MVC或Spring WebFlux控制器或Spring WebFlux功能端点混合使用。在两个框架中支持相同的基于注释的编程模型，可以更轻松地重用知识，同时为正确的工作选择合适的工具。
 *  评估应用程序的一种简单方法是检查其依赖性。如果你要使用阻塞持久性API（JPA，JDBC）或网络API，则Spring MVC至少是常见体系结构的最佳选择。从技术上讲，Reactor和RxJava都可以在单独的线程上执行阻塞调用，但是你不会充分利用非阻塞的Web堆栈。
 *  如果你有一个Spring MVC应用程序调用远程服务，请尝试反应`WebClient`。你可以直接从Spring MVC控制器方法返回反应类型（Reactor，RxJava ）。每次呼叫的延迟或呼叫之间的相互依赖性越大，其益处就越大。Spring MVC控制器也可以调用其他无功组件。
 *  如果你有一个庞大的团队，请记住转向非阻塞，功能和声明性编程时的陡峭学习曲线。在没有完全切换的情况下启动的实用方法是使用反应`WebClient`。除此之外，从小处着手衡量效益。我们希望，对于广泛的应用，这种转变是不必要的。如果你不确定要查找哪些好处，请首先了解非阻塞I / O的工作原理（例如，单线程Node.js上的并发）及其影响。
 *  我认为第三方接口调用最佳，spring Flux 替换 spring mvc 就不做介绍了（只使用WebClient**，**WebSockets）

### **maven** ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-webflux</artifactId>
            </dependency>

### **WebClient** ###

*  Spring WebFlux包含WebClient对HTTP请求的被动，非阻塞。客户端具有功能齐全，流畅的API以及用于声明性组合的反应类型，请参阅 Reactive Libraries。WebFlux客户端和服务器依赖相同的非阻塞编解码器来编码和解码请求和响应内容。 内部WebClient委托给HTTP客户端库。默认情况下，它使用 Reactor Netty，内置支持Jetty 反应式HttpClient，其他可以通过一个插件来插入ClientHttpConnector。简单理解就是没有并发问题。

**使用**

*  最简单方法`WebClient`是通过一个静态工厂方法
 *  `WebClient.create()`
 *  `WebClient.create(String baseUrl)`
 *  上述方法使用Reactor Netty `HttpClient`和默认设置，并期望 `io.projectreactor.netty:reactor-netty`在类路径上。
    
    **你****还可以使用`WebClient.builder()`其他选项：**
 *  `uriBuilderFactory`：自定义`UriBuilderFactory`以用作基本URL。
 *  `defaultHeader`：每个请求的标头。
 *  `defaultCookie`：每个请求的Cookie。
 *  `defaultRequest`：`Consumer`自定义每个请求。
 *  `filter`：每个请求的客户端过滤器。
 *  `exchangeStrategies`：HTTP消息读取器/写入器自定义。
 *  `clientConnector`：HTTP客户端库设置。

ExchangeStrategies strategies = ExchangeStrategies.builder()
                .codecs(configurer -> {
                    // ...
                })
                .build();
    
        WebClient client = WebClient.builder()
                .exchangeStrategies(strategies)
                .build();

**mutate()克隆**

**构建后，`WebClient`实例是不可变的。但是，你****可以克隆它并构建修改后的副本，而不会影响原始实例，如以下示例所示**

WebClient client1 = WebClient.builder()
                .filter(filterA).filter(filterB).build();//Lamda表达式filter获取拦截的对象
    
        WebClient client2 = client1.mutate()
                .filter(filterC).filter(filterD).build();
    
        // client1 has filterA, filterB
    
        // client2 has filterA, filterB, filterC, filterD

**Reactor Netty**

要自定义Reactor Netty设置，只需提供预配置`HttpClient`：

HttpClient httpClient = HttpClient.create().secure(sslSpec -> ...);
    
        WebClient webClient = WebClient.builder()
                .clientConnector(new ReactorClientHttpConnector(httpClient))
                .build();

*  默认情况下，`HttpClient`参与保存的全局Reactor Netty资源 `reactor.netty.http.HttpResources`，包括事件循环线程和连接池。这是推荐的模式，因为固定的共享资源是事件循环并发的首选。在此模式下，全局资源将保持活动状态，直到进程退出。
 *  如果服务器与进程同步，则通常不需要显式关闭。但是，如果服务器可以在进程中启动或停止（例如，部署为WAR的Spring MVC应用程序），则可以`ReactorResourceFactory`使用`globalResources=true`（缺省值）声明Spring管理的bean类型 ，以确保Reactor Netty全局资源是在Spring关闭时`ApplicationContext`关闭，如下例所示：

@Bean
        public ReactorResourceFactory reactorResourceFactory() {
            return new ReactorResourceFactory();
        }

**你还可以选择不参与全局Reactor Netty资源。但是，在此模式下，你****需要确保所有Reactor Netty客户端和服务器实例都使用共享资源，如以下示例所示：**

@Configuration(proxyBeanMethods = false)
    public class WebFluxConfig {
    
        @Bean
        public ReactorResourceFactory resourceFactory() {
            ReactorResourceFactory factory = new ReactorResourceFactory();
            //创建独立于全球资源的资源
            factory.setUseGlobalResources(false);
            return factory;
        }
    
    
        /**
         * TcpClient 配置(HTTP)
         *
         * @return WebClient
         */
        @Bean
        public WebClient webClient() {
           Function<HttpClient, HttpClient> mapper = client -> {
                // 进一步的自定义...
            };
            //将ReactorClientHttpConnector构造函数与资源工厂一起使用
            ClientHttpConnector connector =
                    new ReactorClientHttpConnector(resourceFactory(), mapper);
            //将连接器插入WebClient.Builder
            return WebClient.builder().clientConnector(connector).build();
        }
    }

### retrieve() ###

*  **该`retrieve()`方法是获取响应主体并对其进行解码的最简单方法。以下示例显示了如何执行此操作：**

WebClient client = WebClient.create("https://www.test.com");
    
        Mono<Person> result = client.get()
                .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(Person.class);

*  **获取从响应中解码的对象流，如以下示例所示：**

Flux<Quote> result = client.get()
                .uri("/quotes").accept(MediaType.TEXT_EVENT_STREAM)
                .retrieve()
                .bodyToFlux(Quote.class);

*  **默认情况下，4XX或5xx状态代码的应答导致 `WebClientResponseException`或它的HTTP状态的具体子类之一，比如`WebClientResponseException.BadRequest`，`WebClientResponseException.NotFound`和其他人。你****还可以使用该`onStatus`方法自定义生成的异常，如以下示例所示： **

Mono<Person> result = client.get()
                .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .onStatus(HttpStatus::is4xxClientError, response -> ...)
                .onStatus(HttpStatus::is5xxServerError, response -> ...)
                .bodyToMono(Person.class);

### exchange() ###

*  **该`exchange()`方法提供了比该`retrieve`方法更多的控制。以下示例等同于`retrieve()`但也提供对以下内容的访问`ClientResponse`：**

Mono<Person> result = client.get()
                .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
                .exchange()
                .flatMap(response -> response.bodyToMono(Person.class));

*  **在此级别，你还可以创建一个完整的`ResponseEntity`：**

Mono<ResponseEntity<Person>> result = client.get()
                .uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
                .exchange()
                .flatMap(response -> response.toEntity(Person.class));

*  **请注意（与`retrieve()`）不同，`exchange()`4xx和5xx响应没有自动错误信号。必须检查状态代码并决定如何继续。**

<table> 
 <tbody> 
  <tr> 
   <td><br><span style="color:#000000;">使用时<code>exchange()</code>，必须始终使用任何<code>body</code>或<code>toEntity</code>方法&nbsp;<code>ClientResponse</code>来确保释放资源并避免HTTP连接池的潜在问题。<code>bodyToMono(Void.class)</code>如果没有预期的响应内容，</span>你<span style="color:#000000;">可以使用。但是，如果响应确实包含内容，则连接将关闭，并且不会放回池中。</span></td> 
  </tr> 
 </tbody> 
</table>

### Request Body ###

*  **请求正文可以从一个编码`Object`，如下例所示：**

Mono<Person> personMono = ... ;
    
        Mono<Void> result = client.post()
                .uri("/persons/{id}", id)
                .contentType(MediaType.APPLICATION_JSON)
                .body(personMono, Person.class)
                .retrieve()
                .bodyToMono(Void.class);

*  **你还可以编码对象流，如以下示例所示：**

Flux<Person> personFlux = ... ;
    
        Mono<Void> result = client.post()
                .uri("/persons/{id}", id)
                .contentType(MediaType.APPLICATION_STREAM_JSON)
                .body(personFlux, Person.class)
                .retrieve()
                .bodyToMono(Void.class);

*  **或者，如果你****有实际值，则可以使用syncBody快捷方法，如以下示例所示：**

Person person = ... ;
    
        Mono<Void> result = client.post()
                .uri("/persons/{id}", id)
                .contentType(MediaType.APPLICATION_JSON)
                .syncBody(person)
                .retrieve()
                .bodyToMono(Void.class);

### 表单数据 ###

*  **要发送表单数据，你****可以提供一个`MultiValueMap<String, String>`正文。请注意，内容被自动设置为`application/x-www-form-urlencoded`通过 `FormHttpMessageWriter`。以下示例显示如何使用`MultiValueMap<String, String>`**

MultiValueMap<String, String> formData = ... ;
    
        Mono<Void> result = client.post()
                .uri("/path", id)
                .syncBody(formData)
                .retrieve()
                .bodyToMono(Void.class);

*  **还可以使用`BodyInserters`以下方式提供内联表单数据，如以下示例所示**：

import static org.springframework.web.reactive.function.BodyInserters.*;
    
        Mono<Void> result = client.post()
                .uri("/path", id)
                .body(fromFormData("k1", "v1").with("k2", "v2"))
                .retrieve()
                .bodyToMono(Void.class);

*  **多部分数据**
 *  **要发送多部分数据，你****需要提供`MultiValueMap<String, ?>`其值为`Object`表示部件内容的`HttpEntity`实例或表示部件的内容和标头的实例。`MultipartBodyBuilder`提供了一个方便的API来准备多部分请求。以下示例显示如何创建`MultiValueMap<String, ?>`：**

MultipartBodyBuilder builder = new MultipartBodyBuilder();
        builder.part("fieldPart", "fieldValue");
        builder.part("filePart1", new FileSystemResource("...logo.png"));
        builder.part("jsonPart", new Person("Jason"));
        builder.part("myPart", part); // Part from a server request
    
        MultiValueMap<String, HttpEntity<?>> parts = builder.build();

*  在大多数情况下，你不必`Content-Type`为每个部件指定。内容类型是根据`HttpMessageWriter`所选内容类型自动确定的，以便`Resource`根据文件扩展名自动确定。如有必要，你可以`MediaType`通过其中一个重载的构建器`part`方法显式提供每个部分的使用。
 *  一旦`MultiValueMap`准备好了，`WebClient`通过该`syncBody`方法传递给它的最简单方法是，如下例所示

MultipartBodyBuilder builder = ...;
    
        Mono<Void> result = client.post()
                .uri("/path", id)
                .syncBody(builder.build())
                .retrieve()
                .bodyToMono(Void.class);

*  如果`MultiValueMap`包含至少一个非`String`值，也可以表示常规表单数据（即`application/x-www-form-urlencoded`），则无需将其设置`Content-Type`为`multipart/form-data`。使用时始终如此 `MultipartBodyBuilder`，这确保了`HttpEntity`包装。
 *  作为替代方案`MultipartBodyBuilder`，你还可以通过内置提供内联样式的多部分内容`BodyInserters`，如以下示例所示：

import static org.springframework.web.reactive.function.BodyInserters.*;
    
        Mono<Void> result = client.post()
                .uri("/path", id)
                .body(fromMultipartData("fieldPart", "value").with("filePart", resource))
                .retrieve()
                .bodyToMono(Void.class);

### 客户过滤器 ###

*  **通过注册客户端过滤器（`ExchangeFilterFunction`）`WebClient.Builder` 来拦截和修改请求，如以下示例所示：**

WebClient client = WebClient.builder()
            .filter((request, next) -> {
    
                ClientRequest filtered = ClientRequest.from(request)
                        .header("foo", "bar")
                        .build();
    
                return next.exchange(filtered);
            })
            .build();

*  **这可用于横切关注点，例如身份验证。以下示例使用过滤器通过静态工厂方法进行基本身份验证：**

// static import of ExchangeFilterFunctions.basicAuthentication
    
    WebClient client = WebClient.builder()
            .filter(basicAuthentication("user", "password"))
            .build();

*  **过滤器全局应用于每个请求。要更改特定请求的过滤器行为，你****可以将请求属性添加到`ClientRequest`链中的所有过滤器，然后如下例所示：**

WebClient client = WebClient.builder()
            .filter((request, next) -> {
                Optional<Object> usr = request.attribute("myAttribute");
                // ...
            })
            .build();
    
    client.get().uri("https://example.org/")
            .attribute("myAttribute", "...")
            .retrieve()
            .bodyToMono(Void.class);
    
        }

*  **还可以复制现有`WebClient`，插入新过滤器或删除已注册的过滤器。以下示例在索引0处插入基本身份验证筛选器：**

// static import of ExchangeFilterFunctions.basicAuthentication
    
    WebClient client = webClient.mutate()
            .filters(filterList -> {
                filterList.add(0, basicAuthentication("user", "password"));
            })
            .build();

### 同步使用 ###

*  **`WebClient` 可以通过在结尾处阻塞结果来使用同步样式：**

//block获取结果 bodyToMono单个对象使用，bodyToFlux为集合
    Person person = client.get().uri("/person/{id}", i).retrieve()
        .bodyToMono(Person.class)
        .block();
    
    List<Person> persons = client.get().uri("/persons").retrieve()
        .bodyToFlux(Person.class)
        .collectList()
        .block();

*  **但是，如果需要进行多次调用，则可以更有效地避免单独阻止每个响应，而是等待组合结果**

Mono<Person> personMono = client.get().uri("/person/{id}", personId)
            .retrieve().bodyToMono(Person.class);
    
    Mono<List<Hobby>> hobbiesMono = client.get().uri("/person/{id}/hobbies", personId)
            .retrieve().bodyToFlux(Hobby.class).collectList();
    
    Map<String, Object> data = Mono.zip(personMono, hobbiesMono, (person, hobbies) -> {
                Map<String, String> map = new LinkedHashMap<>();
                map.put("person", personName);
                map.put("hobbies", hobbies);
                return map;
            })
            .block();

*  以上仅是一个例子。有很多其他的模式和运算符可以组合一个反应式管道，它可以进行许多远程调用，可能是一些嵌套的，相互依赖的，直到最后都没有阻塞。

### WebSocket ###

**WebSocket简介（以后的篇章会单独讲解websocket)**

*  WebSocket协议RFC 6455提供了一种标准化方法，通过单个TCP连接在客户端和服务器之间建立全双工双向通信通道。它是来自HTTP的不同TCP协议，但设计为使用端口80和443通过HTTP工作，并允许重用现有的防火墙规则。 WebSocket交互以HTTP请求开始，该HTTP请求使用HTTP Upgrade标头进行升级，或者在这种情况下，切换到WebSocket协议。以下示例显示了这样的交互：

GET /spring-websocket-portfolio/portfolio HTTP/1.1
    Host: localhost:8080
    Upgrade: websocket 
    Connection: Upgrade 
    Sec-WebSocket-Key: Uc9l9TMkWGbHFD2qnFHltg==
    Sec-WebSocket-Protocol: v10.stomp, v11.stomp
    Sec-WebSocket-Version: 13
    Origin: http://localhost:8080

*  具有WebSocket支持的服务器返回类似于以下内容的输出，而不是通常的200状态代码

HTTP/1.1 101 Switching Protocols   #协议切换
    Upgrade: websocket
    Connection: Upgrade
    Sec-WebSocket-Accept: 1qVdfYHU9hPOl4JYYNXF623Gzn0=
    Sec-WebSocket-Protocol: v10.stomp

*  HTTP升级请求下的TCP套接字将保持打开状态，以便客户端和服务器继续发送和接收消息。
 *  如果WebSocket服务器在Web服务器（例如nginx）后面运行，
 *  可能需要将其配置为将WebSocket升级请求传递到WebSocket服务器。

### HTTP与WebSocket ###

*  尽管WebSocket被设计为与HTTP兼容并且以HTTP请求开始，但重要的是要理解这两种协议会导致非常不同的体系结构和应用程序编程模型。
 *  在HTTP和REST中，应用程序被建模为多个URL。要与应用程序交互，客户端访问这些URL，请求 - 响应样式。服务器根据HTTP URL，方法和标头将请求路由到适当的处理程序。
 *  相比之下，在WebSockets中，初始连接通常只有一个URL。随后，所有应用程序消息都在同一TCP连接上流动。这指向完全不同的异步，事件驱动的消息传递体系结构。
 *  WebSocket也是一种低级传输协议，与HTTP不同，它不对消息内容规定任何语义。这意味着除非客户端和服务器就消息语义达成一致，否则无法路由或处理消息。
 *  WebSocket客户端和服务器可以通过`Sec-WebSocket-Protocol`HTTP握手请求中的标头协商使用更高级别的消息传递协议（例如，STOMP）。如果没有，他们需要提出自己的惯例。

### 何时使用WebSocket ###

*  WebSockets可以使网页变得动态和交互。但是，在许多情况下，Ajax和HTTP流式传输或长轮询的组合可以提供简单有效的解决方案。
 *  例如，新闻，邮件和社交订阅源需要动态更新，但每隔几分钟就可以完全正常更新。另一方面，协作，游戏和财务应用程序需要更接近实时。
 *  仅延迟不是决定因素。如果消息量相对较低（例如，监视网络故障），HTTP流式传输或轮询可以提供有效的解决方案。它是低延迟，高频率和高容量的组合，是使用WebSocket的最佳选择。
 *  还要记住，在Internet上，受控制之外的限制性代理可能会妨碍WebSocket交互，因为它们未配置为传递 `Upgrade`标头，或者因为它们关闭看似空闲的长期连接。这意味着将WebSocket用于防火墙内的内部应用程序是一个比面向公众的应用程序更直接的决策。

### WebSocket API ###

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

import org.springframework.web.reactive.socket.WebSocketHandler;
    import org.springframework.web.reactive.socket.WebSocketSession;
    
    public class MyWebSocketHandler implements WebSocketHandler {
    
        @Override
        public Mono<Void> handle(WebSocketSession session) {
            // ...
        }
    }

*  可以将其映射到URL并添加一个`WebSocketHandlerAdapter`，如下例所示：

@Configuration
    static class WebConfig {
    
        @Bean
        public HandlerMapping handlerMapping() {
            Map<String, WebSocketHandler> map = new HashMap<>();
            map.put("/path", new MyWebSocketHandler());
    
            SimpleUrlHandlerMapping mapping = new SimpleUrlHandlerMapping();
            mapping.setUrlMap(map);
            mapping.setOrder(-1); // before annotated controllers
            return mapping;
        }
    
        @Bean
        public WebSocketHandlerAdapter handlerAdapter() {
            return new WebSocketHandlerAdapter();
        }
    }

*  **`WebSocketHandler`**
 *  获取和返回的`handle`方法 指示会话的应用程序处理何时完成。会话通过两个流处理，一个用于入站消息，一个用于出站消息。下表描述了处理流的两种方法：`WebSocketHandler``WebSocketSession``Mono<Void>`

<table> 
 <thead> 
  <tr> 
   <th style="border-color:#cacaca;text-align:left;vertical-align:top;"><code>WebSocketSession</code>&nbsp;方法</th> 
   <th style="border-color:#cacaca;text-align:left;vertical-align:top;">描述</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td style="border-color:#cacaca;text-align:left;vertical-align:top;"> <p style="margin-left:0px;"><span style="color:#000000;"><code>Flux&lt;WebSocketMessage&gt; receive()</code></span></p> </td> 
   <td style="border-color:#cacaca;text-align:left;vertical-align:top;"> <p style="margin-left:0px;"><span style="color:#000000;">提供对入站消息流的访问，并在关闭连接时完成。</span></p> </td> 
  </tr> 
  <tr> 
   <td style="border-color:#cacaca;text-align:left;vertical-align:top;"> <p style="margin-left:0px;"><span style="color:#000000;"><code>Mono&lt;Void&gt; send(Publisher&lt;WebSocketMessage&gt;)</code></span></p> </td> 
   <td style="border-color:#cacaca;text-align:left;vertical-align:top;"> <p style="margin-left:0px;"><span style="color:#000000;">获取传出消息的源，写入消息，并<code>Mono&lt;Void&gt;</code>在源完成并完成写入时返回完成的消息。</span></p> </td> 
  </tr> 
 </tbody> 
</table>

*  一个`WebSocketHandler`必须组成入站和出站流为一个统一的流程，并返回一个`Mono<Void>`反映该流的完成。根据应用程序要求，统一流程在以下情况下完成：
 *  入站或出站消息流完成。
 *  入站流完成（即连接已关闭），而出站流是无限的。
 *  在选定的点上，通过`close`方法`WebSocketSession`。
 *  当入站和出站消息流组合在一起时，无需检查连接是否打开，因为Reactive Streams信号终止活动。入站流接收完成或错误信号，并且出站流接收取消信号。
 *  处理程序的最基本实现是处理入站流的实现。以下示例显示了这样的实现：

class ExampleHandler implements WebSocketHandler {
    
        @Override
        public Mono<Void> handle(WebSocketSession session) {
            return session.receive()    //1         
                    .doOnNext(message -> {
                        // ...             //2     
                    })
                    .concatMap(message -> {
                        // ...          //3        
                    })
                    .then();     //4               
        }
    }

<table> 
 <tbody> 
  <tr> 
   <td>1</td> 
   <td><span style="color:#000000;"><span style="color:#000000;">访问入站流。</span></span></td> 
  </tr> 
  <tr> 
   <td>2</td> 
   <td><span style="color:#000000;"><span style="color:#000000;">对每条消息做点什么。</span></span></td> 
  </tr> 
  <tr> 
   <td>3</td> 
   <td><span style="color:#000000;"><span style="color:#000000;">执行使用消息内容的嵌套异步操作。</span></span></td> 
  </tr> 
  <tr> 
   <td>4</td> 
   <td><span style="color:#000000;"><span style="color:#000000;"><code>Mono&lt;Void&gt;</code>收到完成后返回完成。</span></span></td> 
  </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
  <tr> 
   <td><span style="color:#000000;">对于嵌套的异步操作，您可能需要调用<code>message.retain()</code>使用池化数据缓冲区的底层服务器（例如，Netty）。否则，可能在您有机会读取数据之前释放数据缓冲区。</span></td> 
  </tr> 
 </tbody> 
</table>

*  **以下实现组合了入站和出站流：**

class ExampleHandler implements WebSocketHandler {
    
        @Override
        public Mono<Void> handle(WebSocketSession session) {
    
            Flux<WebSocketMessage> output = session.receive()    //处理入站消息流           
                    .doOnNext(message -> {
                        // ...
                    })
                    .concatMap(message -> {
                        // ...
                    })
                    .map(value -> session.textMessage("Echo " + value));    //创建出站消息，生成组合流
    
            return session.send(output);     //返回我们继续收到时未完成的Mono <Void>                              
        }
    }

*  **入站和出站流可以是独立的，只有完成才能连接，如下例所示：**

class ExampleHandler implements WebSocketHandler {
    
        @Override
        public Mono<Void> handle(WebSocketSession session) {
    
            Mono<Void> input = session.receive()    //处理入站消息流                            
                    .doOnNext(message -> {
                        // ...
                    })
                    .concatMap(message -> {
                        // ...
                    })
                    .then();
    
            Flux<String> source = ... ;
            Mono<Void> output = session.send(source.map(session::textMessage)); //放送传出的消息
    
            return Mono.zip(input, output).then();    //加入流并返回Mono <Void>，当任一流结束时完成                          
        }
    }

*  ### **`DataBuffer`** ###
 *  `DataBuffer`是WebFlux中字节缓冲区的表示。要理解的关键点是在某些服务器（如Netty）上，字节缓冲区被池化并引用计数，并且必须在使用时释放以避免内存泄漏。在Netty上运行时，`DataBufferUtils.retain(dataBuffer)`如果应用程序希望保留输入数据缓冲区以确保它们不被释放，并且随后在使用`DataBufferUtils.release(dataBuffer)`缓冲区时使用，则必须使用它们。

### **服务器配置** ###

*  在`RequestUpgradeStrategy`每个服务器公开可用于底层的WebSocket发动机的WebSocket相关的配置选项。以下示例在Tomcat上运行时设置WebSocket选项：

@Configuration
    static 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);
        }
    }

*  检查服务器的升级策略以查看可用的选项。目前，只有Tomcat和Jetty公开了这样的选项。

### 客户端 ###

*  Spring WebFlux提供了一个`WebSocketClient`抽象，包括Reactor Netty，Tomcat，Jetty，Undertow和标准Java（即JSR-356）的实现。
    
    <table> 
     <tbody> 
      <tr> 
       <td style="text-align:center;vertical-align:top;width:2.8px;">&nbsp;</td> 
       <td><span style="color:#000000;"><span style="color:#000000;">Tomcat客户端实际上是标准Java的扩展，在<code>WebSocketSession</code>处理过程中具有一些额外的功能，以利用特定于Tomcat的API来暂停接收消息以获得反压。</span></span></td> 
      </tr> 
     </tbody> 
    </table>
    
    **要启动WebSocket会话，您可以创建客户端的实例并使用其`execute` 方法：**
 *      WebSocketClient client = new ReactorNettyWebSocketClient();
        
        URI url = new URI("ws://localhost:8080/path");
        client.execute(url, session ->
                session.receive()
                        .doOnNext(System.out::println)
                        .then());
    
    某些客户端（如Jetty）实现`Lifecycle`并需要在使用它们之前停止并启动。所有客户端都具有与底层WebSocket客户端配置相关的构造函数选项。

### **整合** ###

**spring boot 整合flux webClient  (tcp协议\[即常用的HTTP基于TCP协议\])**

*  **无需配置采用默认**

@Getter
    @Setter
    public class City {
        private String status;
        private String province;
    }

public interface CityService {
        City getCity();
    }
    
    
    @Service
    public class CityServiceImpl implements CityService {
    
        public City getCity() {
            //高德测试
            WebClient client = WebClient.create();
            Mono<City> result = client.get()
                    .uri("https://restapi.amap.com/v3/ip").accept(MediaType.APPLICATION_JSON)
                    .headers(httpHeaders -> {
                        httpHeaders.add("output", "JSON");
                        httpHeaders.add("key", "高德key");
                        httpHeaders.add("ip", "百度一个IP地址");
                    })
                    .exchange()
                    .flatMap(response -> response.bodyToMono(City.class));
    
            return result.block();
        }
    
    }

@RunWith(SpringRunner.class)
    @SpringBootTest(classes= WebApplication.class)
    public class Try {
    
        @Autowired
        private CityService cityService;
    
    
        @Test
        public void Test() {
            City city = cityService.getCity();
            System.out.println(city.getProvince());
            System.out.println(city.getStatus());
        }
    }

**spring boot 整合flux WebSocketClient （一般很少使用）**

@Configuration(proxyBeanMethods = false)
    public class WebSocketClientConfig {
    
        @Bean
        public WebSocketHandlerAdapter handlerAdapter() {
            return new WebSocketHandlerAdapter(webSocketService());
        }
    
        @Bean
        public WebSocketService webSocketService() {
            TomcatRequestUpgradeStrategy strategy = new TomcatRequestUpgradeStrategy();
            strategy.setMaxSessionIdleTimeout(0L);
            return new HandshakeWebSocketService(strategy);
        }
    }

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

**下一篇: [RSocket ][RSocket]**

[ReactiveX]: http://reactivex.io/
[RSocket]: https://blog.csdn.net/qq_37059838/article/details/93996753