Spring Boot：如何使用 WebClient 而不是 RestTemplate 来执行非阻塞和异步调用

Question

我有一个使用 Springboot Resttemplate 的 springboot 项目。我们已经从 1.5.3 迁移到 springboot 2.0.1 并且我们正在尝试制作 其余的使用 WebClient 异步调用它。我们曾经使用 Resttemplate 处理接收到的字符串，如下所示。但 WebClient 只返回 Mono 或 Flux 中的数据。如何将数据作为字符串获取。已经尝试过 block() 方法，但它执行异步调用。

@Retryable(maxAttempts = 4, value = java.net.ConnectException.class,
           backoff = @Backoff(delay = 3000, multiplier = 2))
public Mono<String> getResponse(String url) {
    return webClient.get().uri(urlForCurrent).accept(APPLICATION_JSON)
                    .retrieve()
                    .bodyToMono(String.class);
}

使用 RestTemplate 呈现数据流

1. 控制器接收客户端调用
2. 提供者获取String格式的数据
3. 提供者处理字符串
4. 数据提供给控制器

Controller.java

@RequestMapping(value = traffic/, method = RequestMethod.GET,
                produces = MediaType.APPLICATION_JSON_VALUE)
public String getTraffic(@RequestParam("place") String location) throws InterruptedException, ExecutionException {
    String trafficJSON = Provider.getTrafficJSON(location)
    return trafficJSON;
}

Provider.java

public String getTrafficJSON(String location) {
    String url = ----;

    ResponseEntity<String> response = dataFetcher.getResponse(url);

    /// RESPONSEBODY IS READ AS STRING AND IT NEEDS TO BE PROCESSED
    if (null != response {
        return parser.transformJSON(response.getBody(), params);
    }

    return null;
}

DataFetcher.java

@Retryable(maxAttempts = 4,
           value = java.net.ConnectException.class,
           backoff = @Backoff(delay = 3000, multiplier = 2))
public ResponseEntity<String> getResponse(String url) {
    /* ----------------------- */
    return getRestTemplate().getForEntity(urlForCurrent, String.class);
}

Answer 1

由于存在很多误解，所以在这里我要澄清一些事情。

Spring 已正式声明 RestTemplate 位于 maintenence mode 中，因此如果可以，请使用 WebClient，以便尽可能地成为未来的证明。

如RestTemplate API中所述

注意：从 5.0 开始，该类处于维护模式，仅接受较小的更改和错误请求。请考虑使用org.springframework.web.reactive.client.WebClient，它具有更现代的 API 并支持同步、异步和流式处理方案。

非反应式应用程序

如果您的应用程序是非反应式应用程序（不向调用客户端返回通量或单声道），您需要做的就是使用 block() 如果您需要该值。您当然可以在应用程序内部使用Mono 或Flux，但最后您必须调用block() 来获取您需要返回给调用客户端的具体值。

非响应式应用程序使用例如tomcat、undertow 作为底层服务器实现，它遵循 servlet 规范，因此它将为每个请求分配 1 个线程，因此您不会获得响应式应用程序所获得的性能提升。

反应式应用程序

另一方面，如果您有一个响应式应用程序，则在任何情况下都不应该在您的应用程序中调用 block()。阻塞正是​​它所说的，它将阻塞一个线程并阻塞该线程的执行，直到它可以继续前进，这在反应式世界中是不好的。

您也不应该在您的应用程序中调用subscribe除非您的应用程序是响应的最终消费者。例如，如果您正在调用 api 来获取数据并写入应用程序所连接的数据库。您的后端应用程序是最终消费者。如果外部客户端正在调用您的后端（例如 React、Angular 应用程序、移动客户端等），则外部客户端是最终消费者，并且是订阅者。不是你。

这里的底层默认服务器实现是netty 服务器，它是一个非 servlet、基于事件的服务器，不会为每个请求分配一个线程，服务器本身与线程无关，并且任何线程都可用将在任何请求期间随时处理任何事情。

webflux documentation 明确指出，servlet 3.1+ 支持的服务器 tomcat 和 jetty 都可以与 webflux 以及非 serlet 服务器 netty 和 undertow 一起使用。

我怎么知道我有什么应用程序？

Spring 指出，如果您在类路径中同时具有 spring-web 和 spring-webflux，则应用程序将优先使用 spring-web，并且默认情况下会使用底层 Tomcat 服务器启动非反应式应用程序。

如果需要，可以手动覆盖此行为作为弹簧状态。

在应用程序中同时添加 spring-boot-starter-web 和 spring-boot-starter-webflux 模块会导致 Spring Boot 自动配置 Spring MVC，而不是 WebFlux。之所以选择此行为，是因为许多 Spring 开发人员将 spring-boot-starter-webflux 添加到他们的 Spring MVC 应用程序以使用响应式 WebClient。您仍然可以通过将所选应用程序类型设置为 SpringApplication.setWebApplicationType(WebApplicationType.REACTIVE) 来强制执行您的选择。

The “Spring WebFlux Framework”

那么如何按照题主提供的代码来实现WebClient呢？

@Retryable(maxAttempts = 4,
       value = java.net.ConnectException.class,
       backoff = @Backoff(delay = 3000, multiplier = 2))
public ResponseEntity<String> getResponse(String url) {
    return webClient.get()
            .uri(url)
            .exchange()
            .flatMap(response -> response.toEntity(String.class))
            .block();
}

我会说这是最简单且侵入性最小的实现。您当然需要在 @Bean 中构建一个合适的 Web 客户端，并将其自动装配到其类中。

Answer 2

首先要了解的是，如果您需要致电.block()，您不妨坚持使用RestTemplate，使用WebClient 将一无所获。

如果您想从使用 WebClient 中获益，您需要开始以被动的方式思考。反应式过程实际上只是一系列步骤，每个步骤的输入都是之前步骤的输出。当请求进来时，您的代码会创建一系列步骤并立即返回释放 http 线程。然后，当上一步的输入可用时，框架会使用一个工作线程池来执行每个步骤。

这样做的好处是能够以很小的成本重新考虑编写代码的方式，从而极大地提高了接受竞争请求的能力。您的应用程序只需要一个非常小的 http 线程池和另一个非常小的工作线程池。

当您的控制器方法返回Mono 或Flux 时，您已经正确，无需调用block()。

最简单的形式是这样的：

@GetMapping(value = "endpoint", produces = MediaType.TEXT_PLAIN_VALUE)
@ResponseBody
@ResponseStatus(OK)
public Mono<String> controllerMethod() {

    final UriComponentsBuilder builder =
            UriComponentsBuilder.fromHttpUrl("http://base.url/" + "endpoint")
                    .queryParam("param1", "value");

    return webClient
            .get()
            .uri(builder.build().encode().toUri())
            .accept(APPLICATION_JSON_UTF8)
            .retrieve()
            .bodyToMono(String.class)
            .retry(4)
            .doOnError(e -> LOG.error("Boom!", e))
            .map(s -> {

                // This is your transformation step. 
                // Map is synchronous so will run in the thread that processed the response. 
                // Alternatively use flatMap (asynchronous) if the step will be long running. 
                // For example, if it needs to make a call out to the database to do the transformation.

                return s.toLowerCase();
            });
}

转向反应式思维是一个相当大的范式转变，但值得付出努力。坚持下去，一旦您可以在整个应用程序中完全没有阻塞代码，这真的不是那么难。构建步骤并返回它们。然后让框架管理步骤的执行。

如果有任何不清楚的地方，很乐意提供更多指导。

记得玩得开心:)

Answer 3

第一步是用baseUrl构建WebClient对象；

WebClient webClient = WebClient.builder()
    .baseUrl("http://localhost:8080/api") //baseUrl
    .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
    .build();

然后选择方法并附加路径以及请求变量或正文有效负载。

ResponseSpec responseSpec = webClient
    .get()
    .uri(uriBuilder -> uriBuilder.path("/findById") //additional path
        .queryParam("id", id).build())
    .retrieve()
    .onStatus(HttpStatus::is4xxClientError, response -> Mono.error(new CustomRuntimeException("Error")));

等待bodyToMono 的block() 函数的响应。如果您希望响应为字符串，您可以使用 google 的 gson 库对其进行转换。

Object response = responseSpec.bodyToMono(Object.class).block();
Gson gson = new Gson();
String str = gson.toJson(response);

如果你不想知道api调用的状态，你可以这样做。

webClient
    .post()
    .uri(uri -> uri.path("/save").build())
    .body( BodyInserters.fromObject(payload) )
    .exchange().subscribe();

Answer 4

据我了解，您的 Spring Boot Web 应用程序中需要非阻塞/异步和 Web servlet/同步 api 调用。我有完全相同的问题。答案很简单。您需要同时拥有 spring-boot-starter-web 和 spring-boot-starter-webflux 依赖项。 然后，当您要进行异步/非阻塞调用时，请确保按如下方式调用 toFuture() 方法。

configWebClient.post().uri("/alarm/check")
            .body(BodyInserters.fromFormData(formData))
            .exchangeToMono(res -> {
                if (res.statusCode().equals(HttpStatus.OK)) {
                    return res.bodyToMono(Boolean.class);
                } else {
                    return res.createException().flatMap(Mono::error);
                }
            }).toFuture();