使用 Spring Webclient 进行 Scatter & Gather

Question

我是响应式编程概念的新手，我正在尝试构建一个服务，将请求并行发送到两个后端服务并结合这些结果。 这两个后端服务具有不同的响应结构，我创建了一个映射器方法将所有这些转换为一个通用的响应结构。

这是我现在拥有的，当两个服务都返回结果时它正在工作。

public Mono<List<Response>> getRecords(String input){

List<Response> response = new ArrayList<>();

Mono<FirstApiResponse> gResp = this.firstWebClient.get().uri(uriBuilder -> uriBuilder
            .path("/")
            .queryParam("q", input)
            .build()).retrieve()
            .bodyToMono(FirstApiResponse.class).log()
            .timeout(Duration.ofSeconds(50L));

Mono<SecondApiResponse> iResp = this.secondWebClient.get().uri(uriBuilder -> uriBuilder
        .path("/search")
        .queryParam("term", input)
        .build()).retrieve()
        .bodyToMono(SecondApiResponse.class).log().timeout(Duration.ofSeconds(50L));


return Mono.zip(firstResp,secResp).map(objects ->{
    if(firstResp != null)
    response.addAll(Mapper.convert(objects.getT1()));
    if(secResp != null);
    response.addAll(Mapper.convert(objects.getT2()));
    return response;
});

}

public  List<Response> convert(FirstApiResponse resp){
    ////
    Mapping to Response object 
    ////

    return response;
}

public  List<Response> convert(SecondApiResponse resp){
     ////
    Mapping to Response object 
    ////

    return response;
}

我不知道这是否是正确的做法。此外，我想以这样一种方式做到这一点，即如果任何此服务有任何错误，那么它仍应返回其他服务的结果。现在它抛出异常，我无法弄清楚如何正确处理它

如何正确处理这些错误？

Answer 1

这是一个非常有效的场景，有很多方法可以处理它。一种粗略的方法是使用onErrorReturn 一个您可以处理的新模型。它可以是空响应，也可以是模型周围的包装器，无论哪个看起来适合您的场景。

Mono<Wrapper<FirstApiResponse>> gResp = this.firstWebClient.get().uri(uriBuilder -> uriBuilder
     .path("/")
     .queryParam("q", input)
     .build()).retrieve()
     .bodyToMono(FirstApiResponse.class).log()
     .map( response -> new Wrapper().withResponse(response))
     .timeout(Duration.ofSeconds(50L))
     .doOnError(throwable -> logger.error("Failed", throwable))
      .onErrorReturn(new Wrapper().withError( YourDefaultErrorReponse(...));

Mono<SecondApiResponse> iResp = this.secondWebClient.get().uri(uriBuilder -> uriBuilder
    .path("/search")
    .queryParam("term", input)
    .build())
    .retrieve()      
    .bodyToMono(SecondApiResponse.class).log()
    .map( response -> new Wrapper().withResponse(response))
    .timeout(Duration.ofSeconds(50L))
    ..doOnError(throwable -> logger.error("Failed", throwable))
    .onErrorReturn(new Wrapper().withError( YourDefaultErrorReponse(...))

还有一些方法可以返回默认响应。一个简单的方法是使用包装器之类的东西

public final class Wrapper<T> {
  private T response ;
  private Error error;
      
  public Wrapper<T> withResponse ( T response ){
     this.response = response;
     return this;
  }
  public Wrapper<T> withError( Error error) {
     this.error = error;
     return this;
  }

  public Boolean hasError(){
    return error != null ;
  }
      
  public T getResponse(){
   return response;
  }
}