为什么使用 switchIfEmpty 时项目反应器会无限期挂起？答案

【问题标题】：Why does project reactor hang indefinitely when switchIfEmpty is used?为什么使用 switchIfEmpty 时项目反应器会无限期挂起？
【发布时间】：2019-11-21 16:33:14
【问题描述】：

背景

我正在使用 Spring Boot 2.2.1、project-reactor 3.3.0 和 spring-data-mongodb 2.2.1，并且我正在尝试从多个查询中加载数据。我的代码大致是这样的：

Flux.just("type1", "type2", "type3", "type4")
    .concatMap { type ->
        reactiveMongoOperations.find<Map<String, Any>>(BasicQuery("{'type': '$type'}"), "collectionName")
                                .doOnError { e ->
                                    log.error("Caught exception when reading from mongodb: ${e::class.simpleName} - ${e.message}", e)
                                }.switchIfEmpty {
                                    log.warn("Failed to find any documents of type $type")
                                    Mono.empty<Map<String, Any>>()
                                }
    } 
    .. // More operations here
    .subscribe()

问题是如果reactiveMongoOperations.find(..) 没有找到给定类型的任何文档（因此"Failed to find any documents of type $type" 被记录），整个操作将无限期地挂起。如果我删除 switchIfEmpty 子句，操作完成，一切正常。

问题

如果我添加switchIfEmpty 操作，为什么整个操作会挂起？如果我使用flatMap 而不是concatMap 没关系，它最终还是会挂起。
我应该如何记录没有找到特定查询的文档？ IE。我想记录当reactiveMongoOperations.find(..) 返回一个空的Flux 时没有找到任何文档。

【问题讨论】：

您的代码在所有括号中看起来很奇怪，请使用正确的可运行代码更新您的问题
感谢您的建议，但我认为它是正确的，不过它是 Kotlin。我认为 Kotlin 无处不在，但我可能应该改用纯 Java 以使 Java 开发人员更容易。

标签： spring-boot reactive-programming spring-data-mongodb project-reactor

【解决方案1】：

当从 Kotlin 将代码重写为 Java 时（正如 Thomas 在评论中所建议的那样），我找到了答案！我曾假设我使用了 reactor-kotlin-extensions 库提供的 Kotlin reactor.kotlin.core.publisher.switchIfEmpty 扩展函数：

fun <T> Flux<T>.switchIfEmpty(s: () -> Publisher<T>): Flux<T> = this.switchIfEmpty(Flux.defer { s() })

这里不是这种情况，因此我最终使用了Flux 中定义的switchIfEmpty 方法，定义如下：

public final Flux<T> switchIfEmpty(Publisher<? extends T> alternate)

为了让它在没有扩展功能的情况下工作，我可能应该做这样的事情：

.. 
.switchIfEmpty { subscriber ->
    log.warn("Failed to find any documents of type $type")
    subscriber.onComplete()
}

我最初的解决方案不起作用，因为 Java 版本假定我创建一个Publisher（我做了）并且调用这个发布者上的一个函数（我没有）。在 Kotlin 中，lambda 参数是可选的，如果您不需要它，这就是类型系统没有捕捉到这一点的原因。

这是 Kotlin 与 Java 互操作可能比较棘手的一种方式。

【讨论】：