缓存和使缓存的 Mono 无效 [英] Caching and invalidating cached Mono

问题描述

我在尝试缓存 WebClient 返回的 Mono 时遇到问题.代码是这样的:

I'm running into an issue while trying to cache a Mono returned by a WebClient. The code is something like that:

public Mono<Token> authenticate() {
    return cachedTokenMono = cachedTokenMono
        .switchIfEmpty(
            Mono.defer(() -> 
                    getToken())
                    .cache(token ->
                               Duration.between(Instant.now(), token.getExpires().toInstant()),
                           (Throwable throwable) -> Duration.ZERO,
                           () -> Duration.ZERO));
}

目的是缓存用于接收 Token 的 Mono 直到令牌过期.令牌过期后，缓存的 Mono 变为空并请求新令牌.这按预期工作，但不幸的是 switchIfEmpty() 实际上并没有切换"，而是包装了源 Mono.结果，随着越来越多的 SwitchIfEmptyMono 被创建，这会造成内存泄漏.在这种情况下，正确的模式是什么?有没有办法用新的 Mono 替换空的 Mono?

The intention is that the Mono used to receive a Token is cached until the token expires. After the token expires, the cached Mono becomes empty and a new token is requested. This works as expected, but unfortunately switchIfEmpty() does not actually "switch", it wraps the source Mono instead. As a result, this creates a memory leak as more and more wrapping SwitchIfEmptyMono are created. What is the correct pattern in this case? Is there a way to substitute an empty Mono with a new one?

推荐答案

你可以这样做:

private final Mono<Token> authenticateMono = getToken()
            .cache(
                    token -> Duration.between(Instant.now(), token.getExpires().toInstant()),
                    throwable -> Duration.ZERO,
                    () -> Duration.ZERO)

public Mono<Token> authenticate() {
    return authenticateMono;
}

这个想法是您为每次调用 authenticate() 返回相同的缓存 Mono 实例..cache 运算符确保检查每个订阅的缓存结果.

The idea is that you're returning the same caching Mono<Token> instance for each call to authenticate(). The .cache operator ensures that the cached result is checked for each subscription.

特别是:

• 如果新订阅到达并且没有缓存值，则缓存操作符将订阅从 getToken() 返回的 Mono(这将触发令牌检索).
• 如果一个值已被缓存，并且一个新订阅在缓存超时之前到达，那么缓存操作符将把缓存的值发送给新订阅者
• 如果一个值已经被缓存，并且一个新的订阅在缓存超时后到达，那么缓存操作符将重新订阅从返回的MonogetToken()(这将触发令牌重新检索).
• 如果从 getToken() 返回的 Mono 完成时出现异常，则该异常不会被缓存，因此会传播，并且下一个到达的订阅将再次触发令牌检索

• if a new subscription arrives and there is no cached value, then the cache operator will subscribe to the Mono<Token> returned from getToken() (which will trigger the token retrieval).
• if a value has been cached, and a new subscription arrives before the cache has timed out, then the cache operator will emit the cached value to the new subscriber
• if a value has been cached, and a new subscription arrives after the cache has timed out, then the cache operator will resubscribe to the Mono<Token> returned from getToken() (which will trigger the token re-retrieval).
• If the Mono<Token> returned from getToken() completes with an exception, then that exception won't be cached, and therefore will propagate, and the next subscription that arrives will re-trigger the token retrieval again

这一切都假设:

• getToken() 在订阅者到达之前不做任何工作
• getToken() 检索每个订阅者的令牌
• 您只希望所有订阅者使用一个有效令牌

• getToken() does not do any work before a subscriber arrives
• getToken() retrieves the token for each subscriber
• you only want one active token for all subscribers

另请注意，根据您的用例，您可能希望在令牌到期日期之前稍微到期，以解决时钟偏差.即在新令牌实际到期之前抢先检索它，以防止返回在下游有机会使用它之前到期的 Token.

Also note, that depending on your use case, you might want to expire the token slightly before it's expiration date, to account for clock skew. i.e. to preemptively retrieve a new token just before it actually expires to prevent returning a Token that will expire before the downstream has had an opportunity to use it.

这篇关于缓存和使缓存的 Mono 无效的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！