Mongo Flux/Mono 获取对象/s

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。

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

@simon-baslé 的答复更新

 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 Webflux（在后台，Netty）使用的线程来处理应用程序中的每个传入请求。因此，如果您阻止它，您将完全阻止您的应用程序处理新请求。

您的回答，以及一些小的改进

首先，map 可以简化，因为在 Flux 或 Mono 中不允许使用 null 值。如果值不在 DB 中，Spring Data ReactiveCrudRepository 将返回一个空的 Mono，不要与发出 null 的 Mono 混淆：

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

对

---|->
onComplete()

另外，我认为onErrorResume 您打算包装数据库错误，除了“用户存在”的情况？如果是这样，这个onErrorResume 的位置是错误的，因为它类似于catch(Throwable e)，它也会捕获UserByEmailExistException。把它放在map 之前。也可以直接从地图内投掷。

所以这归结为检测一个空的Mono vs 值Mono，将空的替换为异步数据库保存，将值替换为onError：

TL;DR 给我代码

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))
        ;