如何将Mono封装到带超时的外层Mono中实现降级与异步处理

你提供的实现方式完全不符合需求，核心问题有两个：

1. 内层的block()会直接阻塞当前线程等待长耗时任务返回，外层的15秒超时完全无法生效，相当于必须等内层任务跑完或者触发60秒超时，才会继续执行外层逻辑，完全违背了15秒未返回就先返回降级值的要求。
2. 响应式流默认的timeout操作符会在超时时向上游发送取消信号，你如果不做特殊处理，15秒超时触发的同时就会终止长耗时任务的执行，无法满足「无论如何都要让长耗时任务最多运行60秒、成功后仍要异步处理结果」的要求。

正确实现思路 #

要实现需求的核心是把长耗时任务的执行逻辑和返回给调用方的响应流完全解耦，不要让响应流的取消信号传递到长耗时任务上，实现示例如下：

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;
import java.time.Duration;

public class Service {
    // 长耗时方法，泛型Rsp为返回结果类型
    public Mono<Rsp> longRunningProcess() {
        // 原有长耗时逻辑（Web调用、数据库查询等）
    }

    // 生成降级值的方法
    private Rsp getFallbackResult() {
        // 原有降级逻辑
    }

    // 长任务成功后的异步处理逻辑
    private void asyncProcessResult(Rsp rsp) {
        // 原doOnSuccess中的结果处理逻辑
    }

    public Mono<Rsp> processWithFallback() {
        // 1. 定义完整的长耗时任务处理链：60秒超时、成功后异步处理结果、异常自行消化不对外抛出
        Mono<Rsp> longRunningTask = longRunningProcess()
                .timeout(Duration.ofSeconds(60))
                .doOnSuccess(this::asyncProcessResult)
                .onErrorResume(ex -> Mono.empty()) // 长任务报错/超时直接忽略
                .subscribeOn(Schedulers.boundedElastic()) // 扔到独立的IO调度器执行，不占用主流程线程
                .cache(); // 缓存结果避免重复订阅

        // 2. 主动触发长任务订阅，后续主流程的取消信号不会影响长任务运行
        longRunningTask.subscribe();

        // 3. 对外返回的响应流：15秒内长任务返回就用真实结果，超时直接返回降级值
        return longRunningTask
                .timeout(Duration.ofSeconds(15))
                .onErrorResume(ex -> Mono.just(getFallbackResult()));
    }
}

实现说明 #

• 长耗时任务在独立调度器上运行，主动订阅后不会被主流程的取消信号终止，会最多运行60秒，成功就执行异步处理逻辑，超时/报错直接忽略，完全符合需求。
• 全程无block()阻塞调用，符合响应式编程规范，不会造成线程资源浪费。
• 如果长任务是CPU密集型，可以把Schedulers.boundedElastic()替换为Schedulers.parallel()获得更好的性能。

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