如何借助Reactor确保外部REST API至多被触发一次？

你的需求完全可以通过Reactor自带的Mono.cache()实现，它就是专门用来解决“多个订阅者共享同一结果，原逻辑至多执行一次”这类场景的。

核心思路 #

把带条件判断的API调用逻辑封装成一个被缓存的Mono实例，所有需要该API结果的步骤都订阅这个共享实例，而不是在每个步骤里重复创建新的API调用Mono。

具体实现代码 #

首先，提前封装好带条件判断和缓存的API调用逻辑：

// 封装条件判断+API调用，并缓存结果
Mono<Response> sharedApiResponse = Mono.defer(() -> {
    Request request = getRequest();
    // 按你的条件判断是否调用API
    return shouldInvokeApi(request) ? client.invokeExternalApi(request) : Mono.just(null);
}).cache();

然后，所有需要该API结果的流水线步骤，直接复用这个sharedApiResponse即可：

// 步骤1依赖API结果
step1.flatMap(step1Data -> 
    sharedApiResponse.map(apiRes -> combineStep1AndApi(step1Data, apiRes))
);

// 步骤2依赖API结果
step2.flatMap(step2Data -> 
    sharedApiResponse.map(apiRes -> combineStep2AndApi(step2Data, apiRes))
);

为什么这能保证至多调用一次 #

• Mono.defer()的作用是延迟逻辑执行，确保getRequest()和shouldInvokeApi()的判断是在订阅发生时才执行，而不是初始化这个Mono的时候。
• .cache()会缓存第一个订阅触发的执行结果（包括成功响应、失败异常或者空结果），后续所有订阅者都会直接拿到缓存的结果，不会再触发原API调用逻辑。

补充说明 #

• 如果API调用失败，默认的cache()会缓存失败信号，后续订阅也会收到同样的失败，不会重试调用API，完全符合“至多一次”的要求。
• 如果你需要在特定条件下（比如结果过期）重新调用API，可以使用cache()的重载方法，比如cache(Duration.ofMinutes(5))来设置缓存过期时间，但这会打破“至多一次”的限制，根据你的需求选择即可。

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