Flux into Mono List Object - 项目反应堆答案

【问题标题】：Flux into Mono List Object - project reactorFlux into Mono List Object - 项目反应堆
【发布时间】：2020-11-11 12:22:04
【问题描述】：

我正在使用项目反应器，我有下一个问题：

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

我有另一种方法接收客户 ID 并为该客户返回 Flux 付款Flux<PaymentDto>。

这是模型

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:

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

【问题讨论】：

看来付款电话一定是阻塞的，因为客户依赖于付款清单。您是否尝试在不使用 parallelStream() 的情况下运行它？这可能会给你带来问题。也许有一种方法可以重组代码。您可以找到客户，然后在某些服务类中拨打电话以获取付款并将其设置在那里。
@jhon 请检查我的答案。希望能解决你的问题

标签： java reactive-programming spring-webflux project-reactor nonblocking

【解决方案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;
    });
  }

【讨论】：