标签: reactor-netty

我正在使用 Spring WebFlux webclient 进行 REST 调用。我已经以3000毫秒为单位配置了连接超时，因此：

WebClient webClient = WebClient.builder()
    .clientConnector(new ReactorClientHttpConnector(options -> options
        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)))
    .build();

return webClient.get()
    .uri("http://localhost:8081/resource")
    .retrieve()
    .onStatus(HttpStatus::isError, clientResponse -> {
        // Some logging..
        return Mono.empty();
    })
    .bodyToMono(MyPojo.class);

该onStatus方法Mono为每个400/500响应代码返回一个空值。我怎样才能对连接超时甚至读/写超时做同样的事情。现在它只是抛出一个io.netty.channel.ConnectTimeoutException不被处理的onStatus

我的@ExceptionHandler控制器上不需要，因为这些 REST 调用是更复杂流程的一部分，并且通过空Mono元素应该被忽略。

回来spring-web用一个RestTemplate，我记得连接超时也导致了一个RestClientException。所以我们可以捕获RestClientException所有异常和超时。有没有办法我们也可以做到这一点WebClient？

我想在 webfilter 中获取请求正文。

现在，当请求体很大时，

final Flux<DataBuffer> body = exchange.getRequest().getBody();

这个身体有555号，

如何在 webfilter 中获取请求正文？

基本问题：与 RestTemplate 相比，Spring Reactors WebClient 如何实现非阻塞？在将请求分派给外部服务（例如）之后，它是否不必在某处阻塞？HTTP 本质上是同步的，对吗？所以调用应用程序必须等待响应？线程如何知道对来自服务的响应做出反应的上下文？

我对 Spring Webflux 和 WebClient 的使用相当陌生。例如，我所执行的 POST 请求如下：

Mono<ResponseEntity<Resource>> dotResponse = this.webClient
            .method(HttpMethod.POST)
            .uri(new URI("https://test.com/something"))
            .headers(headers -> headers.addAll(requestHeaders))
            .body(BodyInserters.fromResource(resource))
            .exchange()
            .flatMap(response -> response.toEntity(Resource.class));

实际上，代码有点复杂，但这个示例足以演示基本问题。无论如何，这确实有效，并且我得到了想要的结果。现在，由于资源可能相当大，我想使用“Expect：100-continue”标头。

问题是，我现在得到的是一个空的 100 响应，没有任何关于如何触发剩余请求的线索。我在 Google 或 Stack Overflow 上找不到任何关于如何解决这个问题的信息。

如果有任何指点，我将不胜感激。

我正在尝试使用 Springboot-Reactive webclient 进行 HTTP 调用。我的连接因远程服务器错误而关闭。

请找到以下使用Webclient进行休息呼叫的代码。

Mono<String> post(String url, JSONObject body) {

        Mono<String> result = webClient().post().uri(url)
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.APPLICATION_JSON_UTF8)
                .body(BodyInserters.fromObject(body))
                .exchange().log()
                .flatMap { clientResponse ->
                        return clientResponse.bodyToMono(String.class)
                }
        return result
    }

Web客户端创建的代码：

WebClient webClient() {
    TcpClient tcpClient = TcpClient.create()
    .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 30000)          
    .doOnConnected { connection ->
        connection.addHandlerLast(new LoggingHandler(LogLevel.TRACE))
        connection.addHandlerLast(new ReadTimeoutHandler(30))
        .addHandlerLast(new WriteTimeoutHandler(30))
    }
    tcpClient.wiretap(true)

    ReactorClientHttpConnector httpConnector = new ReactorClientHttpConnector(HttpClient.from(tcpClient))


    return WebClient.builder()
        .clientConnector(httpConnector)
        .build()
}

第一次通话后我收到以下日志：

2019-04-10 15:26:31.534  INFO 235344 --- [nio-8080-exec-2] o.a.c.c.C.[Tomcat].[localhost].[/]       : Initializing Spring DispatcherServlet 'dispatcherServlet'
2019-04-10 15:26:31.534  INFO 235344 …

我正在使用 spring-boot-starter-webflux 中的 WebClient。我在生产环境中不断看到来自 Reactor Netty 的对等错误连接重置。然后，reactor netty 在几秒钟（约 10-20 秒）后重试此失败的请求。我没有看到或无法在较低的环境中重现此错误。我无法确定此错误的根本原因，这里我提供了我的 ClientHelper 实现和错误日志，

import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.http.HttpStatus;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.ClientResponse;
import org.springframework.web.reactive.function.client.WebClient;

public class ClientHelper {

  protected WebClient webClient;

  public <T> T post(Object request, TypeReference typeReference,
                    String uri) {
    try {
      ObjectMapper objectMapper = new ObjectMapper();
      String body = objectMapper.writeValueAsString(request);
      ClientResponse clientResponse =
              getWebClient()
                      .post()
                      .uri(uri)
                      .body(BodyInserters.fromObject(body))
                      .exchange()
                      .block();
      return prepareResponse(clientResponse, typeReference, objectMapper);
    } catch (Exception e) {
      return handleException(e);
    }
  }

  protected <T> T prepareResponse(ClientResponse clientResponse, …

我需要查询证书/DNS 名称不匹配的第三方 API。证书中指定的主机名的最左侧标签事先未知（即CN=some-random-hash.example.com），因此我想使用自定义主机名验证器配置 HTTP 客户端。

当使用通过 获取的默认Reactor NettyHttpClient.create()客户端时，HostnameChecker类会失败证书验证并导致抛出异常：

java.security.cert.CertificateException: No subject alternative DNS name matching XXX found.
    at sun.security.util.HostnameChecker.matchDNS(HostnameChecker.java:214)
    at sun.security.util.HostnameChecker.match(HostnameChecker.java:96)
    at sun.security.ssl.X509TrustManagerImpl.checkIdentity(X509TrustManagerImpl.java:455)
    at sun.security.ssl.X509TrustManagerImpl.checkIdentity(X509TrustManagerImpl.java:436)
    at sun.security.ssl.X509TrustManagerImpl.checkTrusted(X509TrustManagerImpl.java:252)
    at sun.security.ssl.X509TrustManagerImpl.checkServerTrusted(X509TrustManagerImpl.java:136)
    at sun.security.ssl.ClientHandshaker.serverCertificate(ClientHandshaker.java:1626)
    ... 30 more

这个相关答案建议按如下方式实现 SNIMatcher：

HttpClient.create()
        .secure(sslContextSpec -> sslContextSpec
                .sslContext(sslContext)
                .handlerConfigurator((handler) -> {
                    SSLEngine engine = handler.engine();
                    SSLParameters params = new SSLParameters();
                    List<SNIMatcher> matchers = new LinkedList<>();
                    SNIMatcher matcher = new SNIMatcher(0) {
                        @Override
                        public boolean matches(SNIServerName serverName) {
                            return true; …

我在将 webflux Web 客户端连接到我的端点时遇到问题。

在我的控制器中，我试图公开 Chocolate DTO 的热流。\n当我尝试使用邮递员调用端点时，它起作用了。

这句话的存在是因为 Stackoverflow 希望我添加更多文本，因为我的问题中有太多代码，对于给您带来的不便，我们深表歉意。

@RestController\npublic class HotChocolateController {\n\n   @GetMapping(value = "/stream/chocolate", produces = MediaType.APPLICATION_STREAM_JSON_VALUE)\n   public Flux<Chocolate> chocolateStream() {\n       return Flux.interval(Duration.ofMillis(500))\n               .map(l -> Chocolate.builder()\n                    .id(String.valueOf(System.currentTimeMillis()))\n                       .name("Hot Chocolate")\n                       .build())\n               .log();\n   }\n}\n

然后我尝试通过单元测试来测试我的设置：

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.DEFINED_PORT)\nclass ReactiveEndpointTests {\n\n    private final WebClient client = WebClient.builder().baseUrl("http://localhost:8080").build();\n\n\n    @Test\n    public void testHotChocolate(){\n        Flux<Chocolate> chocolateSource = client.get().uri("/stream/chocolate").retrieve().bodyToFlux(Chocolate.class);\n        chocolateSource.subscribe(chocolate -> System.out.println());\n    }\n\n

当我运行此测试时，出现以下异常：

reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.NullPointerException\nCaused by: java.lang.NullPointerException: null\n    at reactor.netty.resources.ColocatedEventLoopGroup.register(ColocatedEventLoopGroup.java:71) ~[reactor-netty-0.9.4.RELEASE.jar:0.9.4.RELEASE]\n    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: \nError has …

在我们应用程序的 Kibana 中，我不断看到以下日志行org.springframework.web.reactive.function.client.ExchangeFunctions：

[2f5e234b] Cancel signal (to close connection)

线程是reactor-http-epoll-1这样的。

它可能在两种情况下发生：

• 当连接成功并返回响应时，那么就没有关系了
• 当由于某种未知原因，10 秒后，连接没有返回任何内容，并且此行也会发生，仅此而已。这似乎是一个超时，但我不确定（因为我的 WebClient 配置中的默认超时是 10 秒）

这可能是什么原因造成的？客户端主动掉线还是服务器主动拒绝？

第二种情况是超时吗？但TimeoutException()之后不会被抛出。

我现在doOnCancel()在 WebClient 中进行日志记录来处理第二种情况，但后来我注意到存在情况 1，并且这种doOnCancel()处理不再有意义，因为它似乎在所有情况下都会发生。

我在 Java Spring Boot 框架上，尝试使用 Mono 序列化 Java 对象以供 WebClient 用于发送。我想知道当找到该值时是否可以完全删除该字段null。我似乎没有找到办法做到这一点。试图找到一个注释来看看这是否有效。

下面是一个例子。

我有一个带有对象的 Java 类，如下所示

public class RequestBody {
  private String name_first;
  private String name_last;
  private String email_address;
}

使用构建器模式来构建它。

RequestBody requestBody =
    RequestBody.builder()
        .name_first(input.getName().getFirst())
        .name_last(input.getName().getLast())
        .build();

使用 WebClient + Mono 向另一个 API 进行 RESTful POST

    return requestBodySpec
      .header("Content-Type", "application/json")
      .body(Mono.just(requestBody), RequestBodyClass)
      .retrieve()
      .bodyToMono(String.class)
      .block();

Mono 序列化后的 JSON 结果如下所示。

{
    "name_first": "Foo",
    "name_last": "Bar,
    "email_address": null
}

期望请求 JSON 如下所示。email_address当值为空时完全删除。我们该怎么做呢？

{
    "name_first": "Foo",
    "name_last": "Bar
}