单列表对象项目反应器中的通量

Question

我正在使用project reactor，我有下一个问题：

我有一个返回mono 的方法，它包含一个CustomerDto列表，每个客户机都有属性，其中一个属性是付款列表。但是这个付款清单是空的。

public class CustomerResponse {
    private List<CustomerDto> customers;
}

public class CustomerDto {
    private int id;
    private String fullname;
    private String documentNumber;
    private List<PaymentDto> payments;
}

public interface CustomerService {
    public Mono<CustomerResponse> customerSearch(CustomerRequest request);
}
public interface PaymentService {
    public Flux<PaymentDto> getPayments(int clientId);
}

public Mono<CustomerResponse> getCustomer(CustomerRequest request) {
    return customerService.customerSearch(request).map(resp -> resp.getCustomers())
            .flatMap(customerList -> {
                List<CustomerDto> newCustomerList = customerList.parallelStream().map(customer -> {
                    Flux<PaymentDto> paymentFlux = 
                            paymentService.getPayments(customer.getId());
                    
                    // Here: java.lang.IllegalStateException: block()/blockFirst()/blockLast()
                    customer.setPayments(paymentFlux.collectList().block());
                    return customer;
                }).collect(Collectors.toList());
                
                return Mono.just(new CustomerResponse(newCustomerList));
            });
}

我有下一个例外：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-4
    at reactor.core.publisher.BlockingSingleSubscriber.blockingGet(BlockingSingleSubscriber.java:83) ~[reactor-core-3.3.6.RELEASE.jar:3.3.6.RELEASE]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 

我想知道是否有一种非阻塞或最佳的方法来做这件事

Answer 1

您可以像这样重构代码以避免阻塞调用：

  public Mono<CustomerResponse> getCustomer(CustomerRequest request) {
    Flux<CustomerDto> customerDtoFluxEnriched = customerService.customerSearch(request)
        .map(CustomerResponse::getCustomers).flatMapMany(Flux::fromIterable).flatMap(customerDto -> {
          Flux<PaymentDto> paymentFlux = paymentService.getPayments(customerDto.getId());
          Mono<List<PaymentDto>> paymentListMono = paymentFlux.collectList();
          return paymentListMono.map(paymentList -> {
            customerDto.setPayments(paymentList);
            return customerDto;
          });
        });
    return customerDtoFluxEnriched.collectList().map(customerList -> {
      CustomerResponse customerResponse = new CustomerResponse();
      customerResponse.setCustomers(customerList);
      return customerResponse;
    });
  }