如何区分Reactor中Mono.zip与后续异步操作的错误？

要解决这个问题，核心思路是给不同阶段的错误打上明确标记，或者在不同作用域内隔离错误处理逻辑，下面给你两种实用的实现方式：

方式一：自定义异常类型（最清晰可靠） #

通过定义专属异常类来标记错误来源，后续就能精准区分是Mono.zip阶段失败，还是后续持久化操作失败。

步骤1：定义自定义异常 #

先创建两个异常类，分别对应不同阶段的错误：

// 标记Mono.zip组合失败的异常
class ZipCombinationFailedException extends RuntimeException {
    public ZipCombinationFailedException(Throwable cause) {
        super(cause);
    }
}

// 标记持久化操作失败的异常
class PersistenceOperationFailedException extends RuntimeException {
    public PersistenceOperationFailedException(Throwable cause) {
        super(cause);
    }
}

步骤2：修改业务代码 #

在对应阶段把原始异常包装成自定义异常，最后在统一错误处理时判断类型：

return Mono.zip(mono1, mono2, mono3)
        // 仅处理Mono.zip阶段的错误日志
        .doOnError(err -> loggerUtil.error("Mono.zip多异步任务组合失败: {}", err.getMessage(), err))
        // 将zip阶段的错误包装为自定义异常，标记来源
        .onErrorMap(err -> new ZipCombinationFailedException(err))
        .flatMap(res -> {
            loggerUtil.info("Mono.zip组合成功，开始执行持久化操作");
            // 持久化操作的错误单独处理并包装
            return readPersistenceDAOHandler.upsertOfferInReadCollection(params)
                    .doOnError(err -> loggerUtil.error("持久化更新操作失败: {}", err.getMessage(), err))
                    .onErrorMap(err -> new PersistenceOperationFailedException(err));
        })
        .map(obj -> {
            loggerUtil.info("持久化操作完成，返回结果");
            return obj;
        })
        // 统一处理错误，通过异常类型区分来源
        .doOnError(err -> {
            if (err instanceof ZipCombinationFailedException) {
                loggerUtil.error("最终捕获到【Mono.zip组合】异常", err);
                // 这里可以添加zip失败后的专属处理逻辑
            } else if (err instanceof PersistenceOperationFailedException) {
                loggerUtil.error("最终捕获到【持久化操作】异常", err);
                // 这里可以添加持久化失败后的专属处理逻辑
            } else {
                loggerUtil.error("捕获到未知异常", err);
            }
        })
        .retryWhen(RetryPolicy.retryOnPersistenceFailed(commonConfig.getFailureRetryCount(), commonConfig.getFailureRetryBackoffInMs()));

方式二：作用域隔离处理 #

如果不想新增自定义异常，也可以把后续操作的错误处理完全隔离在flatMap内部，让Mono.zip的错误直接流到外层，通过错误栈或特征信息区分：

return Mono.zip(mono1, mono2, mono3)
        // 仅处理Mono.zip阶段的错误日志
        .doOnError(err -> loggerUtil.error("Mono.zip多异步任务组合失败: {}", err.getMessage(), err))
        .flatMap(res -> {
            loggerUtil.info("Mono.zip组合成功，开始执行持久化操作");
            // 把后续操作的错误处理完全包裹在这个作用域内
            return readPersistenceDAOHandler.upsertOfferInReadCollection(params)
                    .doOnError(err -> loggerUtil.error("持久化更新操作失败: {}", err.getMessage(), err))
                    .map(obj -> {
                        loggerUtil.info("持久化操作完成，返回结果");
                        return obj;
                    });
        })
        // 外层错误处理：这里的错误要么是Mono.zip的，要么是flatMap中未处理的
        .doOnError(err -> {
            // 可以通过异常的栈轨迹、错误消息特征判断来源
            boolean isZipError = err.getStackTrace()[0].getClassName().contains("MonoZip");
            if (isZipError) {
                loggerUtil.error("最终捕获到Mono.zip组合异常", err);
            } else {
                loggerUtil.error("最终捕获到后续操作异常", err);
            }
        })
        .retryWhen(RetryPolicy.retryOnPersistenceFailed(commonConfig.getFailureRetryCount(), commonConfig.getFailureRetryBackoffInMs()));

注意事项 #

• doOnError是副作用操作，不会消费错误，错误会继续向下传递，所以需要用onErrorMap来包装异常，或者用onErrorResume来吞掉错误（如果不需要继续传递的话）。
• 如果你的重试逻辑只想针对持久化失败，记得在重试策略里判断异常类型，避免Mono.zip的错误被重试。

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