java - 使用 Webclient - Webflux 将两个 api 响应合并为一个

Question

我正在使用 WebFlux 和 WebClient，我需要使用两个 API 并合并其响应。
第一个 API 接收类型和文档编号，并返回一个列表，其中包含一个包含客户数据的元素(这就是它的定义方式)。
第二个 API 接收客户 ID 并返回客户付款列表。
我需要使用这两个 API 并返回一个包含客户数据及其付款的实体。
API 客户回复

public class CustomerResponseApi {
    private List<CustomerApi> clientList;
}

public class CustomerApi {
    private int customerId;
    private String documentNumber;
    private String documentType;
    private String firstName;
    private String lastName;
}

API 支付响应

public class PaymentResponseApi {
    private int customerId;
    private LocalDate paymentDate;
    private float amount;
    private String paymentType;
}

最后我应该有这个
CustomerResponse.java

public class CustomerResponse {
    private int customerId;
    private String documentNumber;
    private String documentType;
    private String firstName;
    private String lastName;
    
    private List<PaymentResponseApi> payments;
}

我有一个代理类负责进行 API 调用
CustomerProxy.java

public class CustomerProxy {

    @Value("${api.base-url}")
    private String baseUrl;

    public Mono<CustomerResponseApi> getCustomer(String documentType, String documentNumber) {
        log.info("baseUrl: {}", baseUrl);
        WebClient webClient = WebClient.create(baseUrl);

        return webClient.get()
                .uri(uri -> uri
                        .path("/customers")
                        .queryParam("documentNumber", documentNumber)
                        .queryParam("documentType", documentType)
                        .build()
                )
                .retrieve()
                .bodyToMono(CustomerResponseApi.class);
    }
}

PaymentProxy.java

public class PaymentProxy {

    @Value("${api.base-url}")
    private String baseUrl;

    public Flux<PaymentResponseApi> getCustomerPayment(int customerId) {
        log.info("baseUrl: {}", baseUrl);
        WebClient webClient = WebClient.create(baseUrl);

        return webClient.get()
                .uri(uri -> uri
                        .path("/payments")
                        .queryParam("customerId", customerId)
                        .build()
                )
                .retrieve()
                .bodyToFlux(PaymentResponseApi.class);
    }
}

以及负责合并响应的服务
CustomerServiceImpl.java

public class CustomerServiceImpl implements CustomerService {
    
    @Autowired
    private CustomerProxy customerProxy;
    
    @Autowired
    private PaymentProxy paymentProxy;

    @Override
    public Mono<CustomerResponse> getCustomerAndPayments(String documentType, String documentNumber) {
        return customerProxy.getCustomer(documentType, documentNumber).flatMap(resp -> {
            CustomerApi customerApi = resp.getClientList().get(0); //always returns one customer
            
            // Here is my problem, because getCustomerPayment method returns a Flux
            List<PaymentResponseApi> payments = paymentProxy.getCustomerPayment(customerApi.getCustomerId());
            
            CustomerResponseBuilder customerBuilder = CustomerResponse.builder()
                    .customerId(customerApi.getCustomerId())
                    .documentNumber(customerApi.getDocumentNumber())
                    .documentType(customerApi.getDocumentType())
                    .firstName(customerApi.getFirstName())
                    .lastName(customerApi.getLastName())
                    .payments(payments);
            
            return Mono.just(customerBuilder.build());
        });
    }
}

我该怎么办？

Answer 1

解决这个问题的两种方法:

使用嵌套映射:

public Mono<CustomerResponse> getCustomerAndPayments(String documentType, String documentNumber) {

    return customerProxy.getCustomer(documentType, documentNumber)
        .map(resp -> resp.getClientList().get(0))
        .flatMap(customerApi -> {
          Flux<PaymentResponseApi> paymentProxyFlux = paymentProxy.getCustomerPayment(customerApi.getCustomerId());
          return paymentProxyFlux.collectList()
              .map(payments -> {
                CustomerResponseBuilder customerBuilder = CustomerResponse.builder()
                    .customerId(customerApi.getCustomerId())
                    .documentNumber(customerApi.getDocumentNumber())
                    .documentType(customerApi.getDocumentType())
                    .firstName(customerApi.getFirstName())
                    .lastName(customerApi.getLastName())
                    .payments(payments);
                return customerBuilder.build();
              });
        });


  }

使用 zip :由于您需要来自第二个 API 中第一个 API 响应的信息，因此您需要将这两个链接在一起。 现在，由于它们是异步调用，因此您需要一个 flatMap 或称为 的 flatMap 变体。 flatMapMany 它发出多个元素(这正是您的第二个 API 所做的)。接下来，您需要两个响应来构建您的 CustomerResponse，即您需要 zip 来自两个 API 的两个响应式流响应。

因此基本上使用 Method2 你需要这个:

  public Mono<CustomerResponse> getCustomerAndPayments(String documentType, String documentNumber) {


    Mono<CustomerApi> customerApiMono =  customerProxy.getCustomer(documentType, documentNumber)
        .map(resp -> resp.getClientList().get(0));
    Mono<List<PaymentResponseApi>> paymentResponseApiListMono = customerApiMono
        .flatMapMany(customerApi -> paymentProxy.getCustomerPayment(customerApi.getCustomerId()))
        .collectList();

    return customerApiMono.zipWith(paymentResponseApiListMono)
        .map(tuple -> {
          CustomerApi customerApi = tuple.getT1();
          List<PaymentResponseApi> payments = tuple.getT2();
          CustomerResponseBuilder customerBuilder = CustomerResponse.builder()
              .customerId(customerApi.getCustomerId())
              .documentNumber(customerApi.getDocumentNumber())
              .documentType(customerApi.getDocumentType())
              .firstName(customerApi.getFirstName())
              .lastName(customerApi.getLastName())
              .payments(payments);
          return customerBuilder.build();
        });

  }

Method2的缺点:Api1即客户API将被订阅两次。