如何使用 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. provider 获取 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 modeWebClient如RestTemplate API中所述注意：从 5.0 开始，这个类处于维护模式，只有少量的更改请求和错误被接受。请考虑使用org.springframework.web.reactive.client.WebClient具有更现代 API 并支持同步、异步和流式传输方案的 。非反应性应用如果您的应用程序是非反应性应用程序（不向调用客户端返回通量或单声道），您必须做的是block()在需要该值时使用。您当然可以在您的应用程序内部使用Monoor Flux，但最后您必须调用block()以获取返回给调用客户端所需的具体值。非反应性应用程序使用例如作为底层服务器实现，它遵循 servlet 规范，因此它将为每个请求分配 1 个线程，因此您将无法获得反应性应用程序所获得的性能提升tomcat。undertow响应式应用另一方面，如果您有一个反应性应用程序，则在任何情况下都不应调用block()您的应用程序。阻塞正是它所说的，它会阻塞一个线程并阻止该线程执行直到它可以继续，这在反应世界中是不好的。您也不应该调用subscribe您的应用程序，除非您的应用程序是响应的最终消费者。例如，如果您正在调用一个 api 来获取数据并写入您的应用程序连接到的数据库。您的后端应用程序是最终消费者。如果外部客户端正在调用您的后端（例如反应、角度应用程序、移动客户端等），则外部客户端是最终消费者，并且是订阅者。不是你。这里的底层默认服务器实现是一个netty服务器，它是一个非 servlet、基于事件的服务器，不会为每个请求分配一个线程，服务器本身是线程不可知的，任何可用的线程都将在任何请求期间随时处理任何事情。webflux文档清楚地指出，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)。“Spring WebFlux 框架”那么如何按照题目提供的代码实现WebClient呢？@Retryable(maxAttempts = 4,       value = java.net.ConnectException.class,       backoff = @Backoff(delay = 3000, multiplier = 2))public Mono<String> getResponse(String url) {    return webClient.get()            .uri(url)            .exchange()            .flatMap(response -> response.toEntity(String.class));}我会说这是最简单和最少侵入性的实现。您当然需要构建一个合适的网络客户端，@Bean并将其自动连接到它的类中。

Answer 2

第一步是WebClient使用 baseUrl 构建对象；WebClient&nbsp;webClient&nbsp;=&nbsp;WebClient.builder() &nbsp;&nbsp;&nbsp;&nbsp;.baseUrl("http://localhost:8080/api")&nbsp;//baseUrl &nbsp;&nbsp;&nbsp;&nbsp;.defaultHeader(HttpHeaders.CONTENT_TYPE,&nbsp;MediaType.APPLICATION_JSON_VALUE) &nbsp;&nbsp;&nbsp;&nbsp;.build();然后选择方法并将路径与请求变量或正文有效负载一起附加。ResponseSpec&nbsp;responseSpec&nbsp;=&nbsp;webClient &nbsp;&nbsp;&nbsp;&nbsp;.get() &nbsp;&nbsp;&nbsp;&nbsp;.uri(uriBuilder&nbsp;->&nbsp;uriBuilder.path("/findById")&nbsp;//additional&nbsp;path &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;.queryParam("id",&nbsp;id).build()) &nbsp;&nbsp;&nbsp;&nbsp;.retrieve() &nbsp;&nbsp;&nbsp;&nbsp;.onStatus(HttpStatus::is4xxClientError,&nbsp;response&nbsp;->&nbsp;Mono.error(new&nbsp;CustomRuntimeException("Error")));等待响应的block()功能bodyToMono。如果你想要响应作为字符串，你可以使用谷歌的 gson 库来转换它。Object&nbsp;response&nbsp;=&nbsp;responseSpec.bodyToMono(Object.class).block();Gson&nbsp;gson&nbsp;=&nbsp;new&nbsp;Gson();String&nbsp;str&nbsp;=&nbsp;gson.toJson(response);如果你不想知道 api 调用的状态，你可以像下面那样做。webClient &nbsp;&nbsp;&nbsp;&nbsp;.post() &nbsp;&nbsp;&nbsp;&nbsp;.uri(uri&nbsp;->&nbsp;uri.path("/save").build()) &nbsp;&nbsp;&nbsp;&nbsp;.body(&nbsp;BodyInserters.fromObject(payload)&nbsp;) &nbsp;&nbsp;&nbsp;&nbsp;.exchange().subscribe();

Answer 3

首先要了解的是，如果您需要调用，.block()不妨坚持使用RestTemplate，使用 WebClient 将一无所获。如果你想从使用 WebClient 中获益，你需要开始以反应的方式思考。反应过程实际上只是一系列步骤，每个步骤的输入都是前一步的输出。当收到请求时，您的代码会创建步骤序列并立即返回并释放 http 线程。当上一步的输入可用时，框架然后使用工作线程池来执行每个步骤。这样做的好处是在接受竞争请求的能力上获得了巨大的收益，而代价很小，而不得不重新考虑您编写代码的方式。您的应用程序只需要一个非常小的 http 线程池和另一个非常小的工作线程池。当您的控制器方法返回Monoor时Flux，您就做对了，不需要调用block().像这样的最简单的形式：@GetMapping(value = "endpoint", produces = MediaType.TEXT_PLAIN_VALUE)@ResponseBody@ResponseStatus(OK)public Mono<String> controllerMethod() {&nbsp; &nbsp; final UriComponentsBuilder builder =&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; UriComponentsBuilder.fromHttpUrl("http://base.url/" + "endpoint")&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .queryParam("param1", "value");&nbsp; &nbsp; return webClient&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .get()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .uri(builder.build().encode().toUri())&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .accept(APPLICATION_JSON_UTF8)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .retrieve()&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .bodyToMono(String.class)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .retry(4)&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .doOnError(e -> LOG.error("Boom!", e))&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; .map(s -> {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // This is your transformation step.&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Map is synchronous so will run in the thread that processed the response.&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // Alternatively use flatMap (asynchronous) if the step will be long running.&nbsp;&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; // For example, if it needs to make a call out to the database to do the transformation.&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; return s.toLowerCase();&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; });}转向反应式思考是一个相当大的范式转变，但值得付出努力。坚持住，一旦您能够在整个应用程序中完全没有阻塞代码，这真的不是那么困难。构建步骤并返回它们。然后让框架管理步骤的执行。如果有任何不清楚的地方，我们很乐意提供更多指导。记得玩得开心:)