Reactor Stream异常排查：调用getDeliveryDate无Redis写入无日志输出

问题背景 #

代码逻辑为：getDeliveryDate方法先尝试从Redis获取数据，若不存在则调用getFallbackValue方法；该方法通过WebClient调用API获取数据并转为字符串。但在main方法中调用getDeliveryDate(id, info)后无任何反应——既无数据写入Redis，log.info("value***** {}", value)也无日志输出，已知client.getDeliveryDateInfo(id)返回Mono类型。

代码片段 #

调用代码 #

getDeliveryDate(id, info);

业务处理方法 #

public void getDeliveryDate(String id, Info info) {
    service.getDeliveryDate(id)
           .doOnNext(value -> log.info("value***** {}", value))
           .map(deliveryDateInfo -> info.deliveryDateInfo(deliveryDateInfo))
           .switchIfEmpty(Mono.fromSupplier(() -> {
               itemInfo.setErrorCode("error");
               return info;
           }));
}

Redis查询与回退逻辑 #

public Mono<String> getDeliveryDate(String id) {
    final String key = id;
    return deliveryDateOperations.opsForValue()
                                 .get(key)
                                 .switchIfEmpty(getFallbackValue(id))
                                 .flatMap(value -> deliveryDateOperations.opsForValue()
                                                                         .setIfAbsent(key, value, ttl)
                                                                         .thenReturn(value));
}

远程API调用方法 #

Mono<String> getFallbackValue(String id) {
    return client.getDeliveryDateInfo(id)
                 .map(response -> convertDataToString(response));
}

核心问题原因 #

Reactor（Spring WebFlux基于此框架）的Mono/Flux属于冷流——只有当存在**订阅（subscribe）**操作时，整个数据流才会被触发执行。你的代码存在以下关键问题：

• 在getDeliveryDate方法中，仅构建了Mono的操作链，但未调用subscribe()或类似的订阅方法
• 没有触发订阅，所有后续操作（Redis查询、WebClient调用、日志打印、Redis写入）都不会实际执行

解决方案 #

需要在操作链末尾添加订阅操作，根据场景选择合适的方式：

方式1：直接在方法内调用subscribe()（适用于简单场景） #

修改getDeliveryDate方法，添加订阅触发执行：

public void getDeliveryDate(String id, Info info) {
    service.getDeliveryDate(id)
           .doOnNext(value -> log.info("value***** {}", value))
           .map(deliveryDateInfo -> info.deliveryDateInfo(deliveryDateInfo))
           .switchIfEmpty(Mono.fromSupplier(() -> {
               itemInfo.setErrorCode("error");
               return info;
           }))
           // 添加订阅触发整个数据流执行
           .subscribe();
}

方式2：返回Mono让上层调用者订阅（符合响应式编程规范） #

如果是在响应式环境（如WebFlux接口）中，建议不要在方法内部订阅，而是返回Mono交给上层处理：

public Mono<Info> getDeliveryDate(String id, Info info) {
    return service.getDeliveryDate(id)
                  .doOnNext(value -> log.info("value***** {}", value))
                  .map(deliveryDateInfo -> info.deliveryDateInfo(deliveryDateInfo))
                  .switchIfEmpty(Mono.fromSupplier(() -> {
                      itemInfo.setErrorCode("error");
                      return info;
                  }));
}

调用处添加订阅：

getDeliveryDate(id, info).subscribe();

额外注意事项 #

• 若在Spring WebFlux的Controller中，直接返回Mono即可，框架会自动处理订阅
• 如需处理订阅过程中的异常，可在subscribe中添加错误处理器：.subscribe(result -> {}, error -> log.error("执行失败", error))

内容的提问来源于stack exchange，提问作者george jiang