Java Reactor中switchIfEmpty未执行问题求助

问题根源分析 #

你的代码存在几个关键问题导致switchIfEmpty逻辑未触发，同时还有响应式流使用不当的问题：

• 异步操作未纳入流中：查询数据库后的flatMap里，直接调用updateParamDB但没有将其作为流的一部分返回，Reactor是懒执行模型，这种调用不会实际执行。
• 变量引用错误：多处出现未定义的Param变量，应该使用查询到的item或新创建的实例。
• 流结构逻辑错误：缓存存在时的flatMap如果返回非empty值，会直接跳过后续的switchIfEmpty，但如果缓存存在但updateParamDB返回empty，会错误触发switchIfEmpty，不符合你的需求逻辑。
• 调度器使用冗余：重复添加subscribeOn不会起到预期效果，反而可能导致线程上下文混乱。

修复后的代码示例 #

// 验证缓存
Mono<Param> paramMono = Optional.ofNullable(data.get(id))
        .map(Mono::just)
        .orElseGet(Mono::empty);

return paramMono
        // 缓存存在：执行DB更新并返回结果
        .flatMap(item -> this.updateParamDB(item, nouveauLibelle))
        // 缓存不存在：走DB查询或创建逻辑
        .switchIfEmpty(Mono.defer(() -> 
                paramRepository.findById(id)
                        // DB存在：更新缓存+DB，返回更新后的结果
                        .flatMap(dbItem -> 
                                this.updateParamDB(dbItem, nouveauLibelle)
                                        .doOnNext(updatedItem -> data.put(id, updatedItem))
                        )
                        // DB也不存在：创建新实例并保存
                        .switchIfEmpty(Mono.defer(() -> {
                            if(StringUtils.isNoneBlank(nouveauLibelle)){
                                Param newParam = new Param(); // 替换为实际的实例创建逻辑
                                newParam.setId(id);
                                newParam.setLibelle(nouveauLibelle);
                                return paramRepository.save(newParam)
                                        .doOnNext(savedItem -> data.put(id, savedItem));
                            }
                            // 若nouveauLibelle为空，返回空或默认实例，根据需求调整
                            return Mono.empty();
                        }))
        ))
        // 统一指定订阅线程（如果需要）
        .subscribeOn(scheduler);

关键修复点说明 #

• 异步操作链式调用：将updateParamDB的返回值纳入flatMap的流中，确保操作会被执行；使用doOnNext在操作完成后更新缓存，避免阻塞流的执行。
• 修正变量引用：明确创建新的Param实例，使用查询到的dbItem进行操作，避免未定义变量的问题。
• 明确流分支逻辑：缓存不存在时才进入switchIfEmpty，DB查询不存在时再进入内层switchIfEmpty，逻辑更清晰，符合你的需求。
• 统一调度器：将subscribeOn移到整个流的末尾，确保整个流的订阅操作在指定线程执行，避免冗余。

额外注意事项 #

• 确保updateParamDB方法返回的Mono是正确的：如果更新成功返回更新后的Param实例，失败返回Mono.error或Mono.empty（根据业务需求）。
• 缓存操作data.put要考虑线程安全：如果data是普通的HashMap，在多线程环境下会有并发问题，建议使用ConcurrentHashMap。
• 单元测试正常但运行时异常，可能是因为测试时使用了同步调度器（如Schedulers.immediate()），而运行时使用了异步调度器，要确保所有异步操作都被正确纳入流中，避免遗漏订阅。

内容的提问来源于stack exchange，提问作者Mohammed ESSABRI