如何重新订阅 Mono 并复用请求结果，避免再次调用外部服务？

2026-11-04数据小许

使用同一个经过 cache() 的 Mono

可以用 cache() 缓存外部查询对应的 Mono<String>。收到 Item 时订阅这个 Mono，提前发起查询；后续访问属性时，返回同一个实例。

只要所有访问都使用同一个缓存后的 Mono 实例，查询尚未完成时，后续订阅者会等待同一次查询；查询完成后，后续订阅者会收到缓存的结果。

为什么保存 Optional 不够

原来的 Optional<String> 只能表示当前有没有值，无法区分“请求尚未完成”“请求完成但没有结果”和“请求失败”，也没有通知等待者的机制。异步回调写入普通字段、其他线程读取该字段时，还需要考虑线程可见性。

保存普通的冷 Mono 也不一定能解决问题，因为它通常会在每次订阅时重新执行上游操作。这里可以用 cache() 共享一次执行及其结果。

提前请求，并让后续访问复用结果

下面的示例只保留属性查询相关的代码，通过构造参数传入属性来源：

import reactor.core.publisher.Mono;

public final class Item {
    private final Mono<String> attributeMono;

    public Item(Mono<String> attributeSource) {
        this.attributeMono = attributeSource.cache();

        // 提前启动查询，并处理这个预取订阅收到的错误。
        this.attributeMono.subscribe(
            ignored -> { },
            error -> System.err.println("属性预取失败: " + error)
        );
    }

    public Mono<String> getAttribute() {
        return attributeMono;
    }
}

创建对象时，可以用 Mono.defer 将外部服务方法的调用推迟到首次订阅：

Item item = new Item(
    Mono.defer(() -> externalService.methodThatReturnsMono(id))
);

构造方法里的预取订阅会触发查询。后续业务通过响应式链使用属性：

Mono<String> result = item.getAttribute()
    .map(attribute -> "属性值: " + attribute);

订阅这条链时，如果属性还没返回，map 会在结果到达后执行，无需轮询或阻塞当前线程。预取订阅的错误处理只负责该订阅；缓存的错误仍会传递给后续订阅者。

下面的写法无法跨调用共享缓存：

public Mono<String> getAttribute() {
    return externalService.methodThatReturnsMono(id).cache();
}

每次调用这个方法都会创建新的缓存实例。前面的示例将缓存后的 Mono 保存在字段中，可以供多次调用复用。

如果必须保留同步返回方式

如果方法必须直接返回 String，结果尚未就绪时就需要阻塞等待；另一种选择是改变接口约定。允许阻塞的调用位置可以提供同步适配方法：

public String getAttributeBlocking() {
    return attributeMono.block();
}

这里的 block() 会订阅同一个缓存 Mono，因此可以复用预取请求。若源以空结果完成，返回值是 null；若源失败，调用方会收到异常。

不要在 Reactor 的非阻塞线程或 WebFlux 请求处理链中使用这种阻塞方式。响应式流程应继续返回 Mono<String>，由调用方通过 map、flatMap 等操作组合后续逻辑。

使用时需要注意什么

无参数的 cache() 会缓存值、空完成或错误。一次查询失败后，后续订阅也会收到缓存的错误；如果需要重试，应另行设计重试策略及其与缓存的顺序。
缓存只在这个 Mono 实例内有效。即使两个 Item 使用相同的 id，它们也不会自动共享请求。
subscribe() 本身不会自动切换线程。如果外部服务方法或其上游执行阻塞操作，预取订阅仍可能阻塞构造过程，需要根据实际调用方式安排执行线程。
如果不要求收到 Item 后立即查询，可以去掉构造方法中的订阅，让第一次业务订阅触发请求，后续订阅仍复用缓存。