spring-webflux - 从Spring WebFlux Web应用程序中的WebClient调用中缓存Mono的结果

Question

我希望缓存Mono(仅在成功的情况下)，这是WebClient调用的结果。

通过阅读项目 react 堆插件文档，我觉得CacheMono不太适合，因为它也存储了我不想要的错误。

因此，我没有使用CacheMono而是执行以下操作:

Cache<MyRequestObject, Mono<MyResponseObject>> myCaffeineCache = 
    Caffeine.newBuilder()
            .maximumSize(100)
            .expireAfterWrite(Duration.ofSeconds(60))
            .build();

MyRequestObject myRequestObject = ...;

Mono<MyResponseObject> myResponseObject = myCaffeineCache.get(myRequestObject,
    requestAsKey -> WebClient.create()
                             .post()
                             .uri("http://www.example.com")
                             .syncBody(requestAsKey)
                             .retrieve()
                             .bodyToMono(MyResponseObject.class)
                             .cache()
                             .doOnError(t -> myCaffeineCache.invalidate(requestAsKey)));

在这里，我在 Mono上调用缓存，然后将其添加到咖啡因缓存中。

任何错误都将输入 doOnError以使缓存无效。

这是缓存 Mono WebClient响应的有效方法吗？

Answer 1

这是极少数的用例之一，实际上允许您调用非 react 性库并将其包装为 react 性类型，并在诸如doOnXYZ的副作用运算符中进行处理，因为:

咖啡因是内存中的缓存，据我所知，不涉及任何I/O

缓存通常不提供有关缓存值的有力保证(这很“容易被人遗忘”)

然后，在这种情况下，您可以查询缓存以查看是否有缓存版本(将其包装并立即返回)，并在 doOn运算符中缓存成功的真实响应，如下所示:

public class MyService {

    private WebClient client;

    private Cache<MyRequestObject, MyResponseObject> myCaffeineCache;

    public MyService() {
        this.client = WebClient.create();
        this.myCaffeineCache = Caffeine.newBuilder().maximumSize(100)
          .expireAfterWrite(Duration.ofSeconds(60)).build();
    }

    public Mono<MyResponseObject> fetchResponse(MyRequestObject request) {

        MyResponseObject cachedVersion = this.myCaffeineCache.get(myRequestObject);
        if (cachedVersion != null) {
           return Mono.just(cachedVersion);
        } else {
           return this.client.post()
                         .uri("http://www.example.com")
                         .syncBody(request.getKey())
                         .retrieve()
                         .bodyToMono(MyResponseObject.class)
                         .doOnNext(response -> this.myCaffeineCache.put(request.getKey(), response));
    }
}

请注意，我不会在此处缓存响应式(Reactive)类型，因为一旦值由缓存返回，就不会涉及I/O或背压。相反，订阅和其他响应流限制使事情变得更加困难。

另外，您对 cache运算符是正确的，因为它不是缓存本身的值，而是更多有关重播其他订户发生的事情。我相信 cache和 replay运算符实际上是 Flux的同义词。