Flux嵌套端点调用返回空列表及状态异常问题排查

核心问题 #

你的代码返回空列表和初始NOT_QUERIED的根本原因是误用subscribe()触发异步操作，主线程未等待异步任务完成就提前返回了初始化变量。

subscribe()是异步触发响应式流执行的方法——调用后主线程会立刻继续执行后续代码，此时四个subscribe内部的处理逻辑还没启动（或刚启动），所以返回的apps是空列表，四个SuccessEnum变量仍保持初始值。另外，用可变集合和可变变量存储结果的方式，既不符合响应式编程的无状态理念，还存在线程安全风险。

正确的响应式实现 #

使用Reactor的zip操作符等待所有异步请求完成，再统一组合结果，全程用响应式操作符替代手动订阅：

fun getApps(id: Long): Flux<Response> {
    // 封装每个请求+处理逻辑为Mono，返回状态与Apps列表的结果对
    val app1Mono = fetchApps1Response(id)
        .collectList()
        .map { processApps(it, id) }
        .onErrorReturn(SuccessEnum.FAILED to emptyList()) // 可选：处理单个请求失败场景

    val app2Mono = fetchApps2Response(id)
        .collectList()
        .map { processApps(it, id) }
        .onErrorReturn(SuccessEnum.FAILED to emptyList())

    val app3Mono = fetchApps3Response(id)
        .collectList()
        .map { processApps(it, id) }
        .onErrorReturn(SuccessEnum.FAILED to emptyList())

    val app4Mono = fetchApps4Response(id)
        .collectList()
        .map { processApps(it, id) }
        .onErrorReturn(SuccessEnum.FAILED to emptyList())

    // 等待所有Mono完成，组合结果生成最终Response
    return Mono.zip(app1Mono, app2Mono, app3Mono, app4Mono)
        .map { tuple ->
            val (app1Result, app2Result, app3Result, app4Result) = tuple
            // 合并所有Apps列表
            val combinedApps = buildList {
                addAll(app1Result.second)
                addAll(app2Result.second)
                addAll(app3Result.second)
                addAll(app4Result.second)
            }
            Response(
                app1Success = app1Result.first,
                app2Success = app2Result.first,
                app3Success = app3Result.first,
                app4Success = app4Result.first,
                apps = combinedApps
            )
        }
        .flux() // 转换为Flux<Response>匹配方法返回值
}

关键说明 #

• Mono.zip的作用：会等待所有传入的Mono都完成后，将它们的结果组合成一个Tuple，确保所有异步请求处理完成后再生成最终响应。
• 错误处理：添加onErrorReturn可以保证单个请求失败时，整个流不会中断，而是返回预设的失败状态和空列表，避免影响其他请求的结果。
• 无状态编程：全程使用不可变集合（buildList创建的列表）和响应式操作符，避免了线程安全问题，符合Reactor的编程模型。

内容的提问来源于stack exchange，提问作者M F