Reactor中如何判断Mono<Void>是否实际被执行？

问题根源 #

Reactor 中，经过filter后得到的空Mono会被判定为正常成功终止，因此链末尾的doOnSuccess无论过滤是否通过都会执行——它监听的是整个序列的完成信号，而非service.loadSmth的执行结果。

无需修改方法签名的解决方案 #

核心思路是：将serviceContinueWork.doSmthElse()的触发逻辑，绑定到service.loadSmth()的执行链上，而非整个主序列的成功信号。

方案一：将doOnSuccess移至flatMap内部

直接在flatMap中为service.loadSmth()的成功结果绑定触发逻辑：

public Mono<Void> execute(String resource) {
    return Mono.just(resource)
            .filter(Objects::nonNull)
            .filter(msg -> msg.startsWith("http://"))
            // 仅当过滤通过时，才会进入此flatMap执行loadSmth
            .flatMap(msg -> service.loadSmth(msg)
                    // 仅在loadSmth成功完成时触发后续逻辑
                    .doOnSuccess(v -> serviceContinueWork.doSmthElse()))
            .doOnError(t -> log.error("service#loadSmth error: {}", ExceptionUtils.getStackTrace(t)))
            .then();
}

原理：只有当消息通过两层过滤后，才会进入flatMap执行service.loadSmth()，此时的doOnSuccess仅监听该loadSmth的成功信号，过滤未通过时完全不会触发。

方案二：用then串联同步逻辑（适合doSmthElse为无返回同步操作）

如果serviceContinueWork.doSmthElse()是同步无返回的操作，也可以用then(Mono.fromRunnable(...))来串联：

public Mono<Void> execute(String resource) {
    return Mono.just(resource)
            .filter(Objects::nonNull)
            .filter(msg -> msg.startsWith("http://"))
            .flatMap(msg -> service.loadSmth(msg)
                    // loadSmth成功后执行同步逻辑
                    .then(Mono.fromRunnable(() -> serviceContinueWork.doSmthElse())))
            .doOnError(t -> log.error("service#loadSmth error: {}", ExceptionUtils.getStackTrace(t)))
            .then();
}

原理：then会等待loadSmth的Mono<Void>成功完成后，再执行Mono.fromRunnable中的逻辑，同样仅在过滤通过且loadSmth成功时触发。

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