为什么不能多次订阅发布者？

Question

我写了两个方法，第一个方法仅调用发布者（productDtoMono）一次，另一个方法调用它两次， 从控制器调用时，第一个返回错误请求（空主体） 而第二个工作完美并返回 http ok 我做了功课，发现我不能多次使用出版商 这是为什么 ？

public Mono<ProductDto> firstupdateProduct(Mono<ProductDto> productDtoMono) {
    return productDtoMono.flatMap(productDto ->
            productRepository.findById(productDto.getId())
                    .flatMap(p -> productDtoMono
                            .map(EntityDtoUtil::toEntity)))
                    .flatMap(productRepository::save)
                    .map(EntityDtoUtil::toDto);
}
public Mono<ProductDto> secondupdateProduct(String id, Mono<ProductDto> productDtoMono) {
    return productRepository.findById(id)
            .flatMap(product -> productDtoMono
                    .map(EntityDtoUtil::toEntity))
            .flatMap(productRepository::save)
            .map(EntityDtoUtil::toDto);
}

我的控制器方法：

@PutMapping
public Mono<ResponseEntity<ProductDto>> updateProduct(@RequestBody Mono<ProductDto> productDtoMono){
    System.out.println("in controller "+productDtoMono);
    return productService.updateProduct(productDtoMono)
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());

}

Answer 1

我做了更多研究。 Mono、Flux默认是冷发布者，因此当它们的创建来源来自请求时，它们只能被消费一次。当我们尝试第二次使用它时，它会尝试从请求生成对象，因此会抛出错误的请求。我必须在将主体传递给服务时对其进行缓存，如下所示：

@PutMapping
public Mono<ResponseEntity<ProductDto>> updateProduct(@RequestBody Mono<ProductDto> productDtoMono){
    System.out.println("in controller "+productDtoMono);
    return productService.updateProduct(productDtoMono.cache())
            .map(ResponseEntity::ok)
            .defaultIfEmpty(ResponseEntity.notFound().build());

}

并像这样更改了 updateProduct

public Mono<ProductDto> updateProduct(Mono<ProductDto> productDtoMono) {
    return productDtoMono.map(ProductDto::getId)
            .flatMap(productRepository::findById)
            .flatMap(product -> productDtoMono.map(EntityDtoUtil::toEntity))
            .flatMap(productRepository::save)
            .map(EntityDtoUtil::toDto);
}