如何在Mono/Flux中调用WebClient方法？实践疑问与示例需求

核心问题 #

你当前的代码存在两个关键错误：

• 在map操作里调用subscribe()：map是同步转换操作符，只能处理纯数据转换，不能触发异步操作。手动调用subscribe()会直接启动WebClient请求，脱离原响应式调用链，导致方法返回的Mono变成Mono<Disposable>（订阅的返回值），上层调用者无法感知请求结果、错误，完全破坏了响应式流的连贯性。
• 用map处理异步操作：map不支持返回Mono/Flux类型，必须用flatMap包裹异步操作，它会把内部的异步流合并到原流中，保持整个调用链的响应式特性。

修正后的代码示例 #

把map替换为flatMap，并移除subscribe()，让整个流保持响应式：

return mailTemplateMappingRepository
    .findById(request.getTemplateKey())
    .switchIfEmpty(Mono.error(new MailTemplateNotSupportedException(
        "The template with key " + request.getTemplateKey() + " is not supported!!!")))
    .flatMap(t -> { // 替换为flatMap
        log.info("sendEmailWithRetry: request {}", request);
        log.info("sendEmailWithRetry: templateMappings {}", t);

        if (!businessUnitAuthTokens.containsKey(t.getExactTargetBusinessUnit())) {
            updateBusinessUnitToken(t);
            // 注意：如果updateBusinessUnitToken是阻塞操作，要改成响应式（比如返回Mono<Void>），然后用flatMap链式调用
            // 示例：return updateBusinessUnitToken(t).then(Mono.just(t));
        }

        String token = "Bearer " + businessUnitAuthTokens.get(t.getExactTargetBusinessUnit());
        String uri = exactTargetMessageDefinitionSendsUrl.replace("{key}", t.getExactTargetKey());
        Map<String, Object> mailTriggerPayload = generateMailTriggerPayload(request);

        RetryBackoffSpec is401RetrySpec = Retry.backoff(1, Duration.ofSeconds(2))
            .filter(throwable -> throwable instanceof Unauthorized)
            .doBeforeRetry(retrySignal -> {
                log.error("UNAUTHORIZED... WILL TRY AGAIN... Request: " + request);
                updateBusinessUnitToken(t);
                // 同样，这里如果update是阻塞操作，要改成响应式，否则会阻塞事件循环
            })
            .onRetryExhaustedThrow((retryBackoffSpec, retrySignal) ->
                new ExactTargetException(HttpStatus.UNAUTHORIZED.value(), retrySignal.failure().getMessage(),
                    "There is something wrong with the authorization key for business unit: "
                        + request.getTemplateKey())
            );

        return restClientService.post(uri, mailTriggerPayload, token, String.class)
            .onErrorResume(error -> {
                if (error instanceof ExactTargetException) {
                    return Mono.error(new ExactTargetException(((ExactTargetException) error).getStatus(),
                        ((ExactTargetException) error).getBody(),
                        "Exact Target has errors... Status: " + ((ExactTargetException) error).getStatus()
                            + " Entity: "
                            + ((ExactTargetException) error).getBody()));
                }
                return Mono.error(error);
            }).retryWhen(is401RetrySpec); // 移除subscribe()
    });

关键注意事项 #

1. 禁止在非终端操作符内调用subscribe()：响应式流的订阅应该由上层调用者负责（比如WebFlux控制器、Spring Batch响应式任务等），手动订阅会导致流脱离控制，无法统一处理错误和结果。
2. 区分map和flatMap：
   • map：同步转换，输入T输出R，适合简单数据转换。
   • flatMap：异步转换，输入T输出Mono<R>/Flux<R>，适合处理数据库查询、HTTP请求这类异步操作。
3. 阻塞操作的响应式改造：如果updateBusinessUnitToken是阻塞方法（比如调用同步数据库接口），必须把它包装成响应式的，避免阻塞Netty事件循环。示例：

   private Mono<Void> updateBusinessUnitToken(MailTemplateMapping t) {
       return Mono.fromCallable(() -> {
           // 原阻塞逻辑
           return null;
       }).subscribeOn(Schedulers.boundedElastic()); // 切换到阻塞线程池
   }
   

相关学习参考 #

• Spring官方响应式操作符文档：重点关注flatMap、switchMap等异步操作符的用法，以及响应式流的订阅机制。
• WebClient官方指南：里面有大量用flatMap结合WebClient处理链式请求的示例，能帮助理解如何保持响应式调用链的连贯性。
• 响应式编程核心原理：理解“冷流”“热流”的区别，以及订阅触发执行的机制，从根本上避免手动订阅的错误。

内容的提问来源于stack exchange，提问作者Александр Хворостенко