Mongo Flux/Mono get对象

Question

我想从Flux/Mono中获取对象。我使用

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>

我会这样做：

    Mono<UserEntity> byEmail = userRepository.findByEmail(userDto.getEmail());
    UserEntity block = byEmail.block();

我有错误：

java.lang.IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread reactor-http-nio-3

为什么？有什么不同的方法来获取对象？

在反应式编程中，如何做到：在RequestBody中，您有UserDto。

如果不创建用户，请检查数据库中是否存在电子邮件。

 return userRepository.findByEmail(userDto.getEmail())
    //rethrow DB errors as BadRequestException, except user exists
    .onErrorResume(t -> Mono.error(new BadRequestException()))
    //throwing in map is converted to onError signal
    .map(ifItGoesThereItExist-> { throw new UserByEmailExistException(); })
    //switchIfEmpty to replace an empty source with a different Mono
    .switchIfEmpty(createUser(userDto))
    //at this point, map can only receive an onNext from the createUser
    .map(u -> new ResponseEntity(u, HttpStatus.CREATED))
    ;

Answer 1

你似乎已经找到了一个答案，但让我详细说明一下。

为什么你不能阻挡

您收到的错误消息表明您尝试在不适应阻塞的特定Thread（或线程池）中恢复阻塞行为。这是Spring Webflow（以及在封面下的Netty）用于处理应用程序中的每个传入请求的线程。因此，如果您阻止它，您将完全阻止您的应用程序提供新请求。

您的答案和一些小的改进

首先，可以简化映射，因为通量或单声道中不允许有空值。如果该值不以DB为单位，Spring数据的反应CRUDepository将返回一个空的单声道，不要与发出null的单声道混淆：

---( null )---|->
onNext(null), onComplete() // onNext(null) being forbidden

vs公司

---|->
onComplete()

此外，我认为使用onErrorResume，除了“用户存在”情况外，您打算包装DB错误？如果是这样，则此错误恢复的位置是错误的，因为它类似于捕获（抛出e），也将捕获UserByEmailExistException。把它放在地图的前面。您也可以从地图内直接投掷。

因此，这归结为检测空的单声道与有值的单声道，将空的单声道替换为异步DB存储，将有值的单声道替换为错误：

太长别读密码

return userRepository.findByEmail(userDto.getEmail())
        //rethrow DB errors as BadRequestException, except user exists
        .onErrorResume(t -> Mono.error(new BadRequestException()))
        //throwing in map is converted to onError signal
        .map(ifItGoesThereItExist-> { throw new UserByEmailExistException(); })
        //switchIfEmpty to replace an empty source with a different Mono
        .switchIfEmpty(createUser(userDto))
        //at this point, map can only receive an onNext from the createUser
        .map(u -> new ResponseEntity(u, HttpStatus.CREATED))
        ;