使用 Spring WebClient 反复过滤响应答案

【问题标题】：Repeatedly filter a response using Spring WebClient使用 Spring WebClient 反复过滤响应
【发布时间】：2020-11-24 11:05:02
【问题描述】：

我是 Spring 新手，甚至是 WebClient 新手。我想使用 Springs 的 WebClient 以 1 秒的间隔重复过滤 Get 响应的主体，持续 2 分钟。我正在执行一个 get 请求，它返回一个空的 JSON 字符串列表。在某个时刻，正文将被填充，我想返回这个字符串列表。我想以这样的方式过滤响应，当它为空时，它会继续执行请求，直到它被填充并返回所需的结果。

 private List<String> checkUser() {
        List<String> ibanList = new ArrayList<>();

        ExchangeFilterFunction filter = ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
            if (clientResponse.body())
                  //something here

        });

        Optional<Account[]> accountsOptional = webClient.get()
                .uri("example.com")
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToMono(Account[].class)
                .delaySubscription(Duration.ofSeconds(1))
                .retry()
                .filter(filter)
                .blockOptional(Duration.ofMinutes(2));

        if (accountsOptional.isPresent()) {

            for (Account account : accountsOptional.get()) {
                ibanList.add(account.getIban());
            }
            return ibanList;
        }
        return null;
    }

有人知道怎么做吗？任何帮助将不胜感激。

【问题讨论】：

标签： java spring spring-boot webclient spring-webflux

【解决方案1】：

您可以使用 bodyToFlux 而不是使用 bodyToMono，然后使用间隔方法。您可以像这样分离出请求：

Flux<Account[]> request = webClient.get()
                .uri("example.com")
                .accept(MediaType.APPLICATION_JSON)
                .retrieve()
                .bodyToFlux(Account[].class);

然后使用间隔调用

Flux.interval(Duration.ofSeconds(1), Duration.ofMinutes(2))
  .map(i -> request)...

然后您可以将过滤器逻辑链接到此

【讨论】：

感谢您的回复！困扰我的是确切的过滤器逻辑。你知道我该怎么做吗？

【解决方案2】：

对我来说，诀窍是在列表为空时使用 flatMap 引发异常，然后调用 retryWhen，直到列表被填充。（blockOptional 已移除，因为不再需要）

Account[] accounts = webClient.get()
                    .uri("example.com")
                    .accept(MediaType.APPLICATION_JSON)
                    .retrieve()
                    .bodyToMono(Account[].class)
                    .delaySubscription(Duration.ofSeconds(1))
                    .flatMap(resp -> {
                        if (resp.length == 0) {
                            return Mono.error(Exception::new);
                        } else {
                            return Mono.just(resp);
                        }
                    })
                    .retryWhen(Retry.max(60))
                    .block();

【讨论】：