执行Spring Web流量中没有人订阅的Mono流

Question

我有一个springWebflux应用程序。此应用程序有两个重要部分：

1. 作业计划以固定间隔运行
2. 作业从DB中获取数据，并将数据存储在Redis中

void run() {
redisAdapter.getTtl()
.doOnError(RefreshExternalCache::logError)
.switchIfEmpty(Mono.defer(() -> {
log.debug(">> RefreshExternalCache > refreshExternalCacheIfNeeded => Remaining TTL could not be retrieved. Cache does not exist. " +
"Trying to create the cache.");
return Mono.just(Duration.ofSeconds(0));
}))
.subscribe(remainingTtl -> {
log.debug(">> RefreshExternalCache > refreshExternalCacheIfNeeded => original ttl for the cache: {} | ttl for cache in seconds = {} | ttl for cache in minutes = {}",
remainingTtl, remainingTtl.getSeconds(), remainingTtl.toMinutes());
if (isExternalCacheRefreshNeeded(remainingTtl, offerServiceProperties.getExternalCacheExpiration(), offerServiceProperties.getExternalCacheRefreshPeriod())) {
log.debug(">> RefreshExternalCache > refreshExternalCacheIfNeeded => external cache is up-to-date, skipping refresh");
} else {
log.debug(">> RefreshExternalCache > refreshExternalCacheIfNeeded => external cache is outdated, updating the external cache");
offerService.refreshExternalCache();
}
});
}

这基本上调用了另一种称为refreshExternalCache()的方法，实现如下：

public void refreshExternalCache() {
fetchOffersFromSource()
.doOnNext(offerData -> {
log.debug(LOG_REFRESH_CACHE + "Updating local offer cache with data from source");
localCache.put(OFFER_DATA_KEY, offerData);
storeOffersInExternalCache(offerData, offerServiceProperties.getExternalCacheExpiration());
})
.doOnSuccess(offerData -> meterRegistry.counter(METRIC_EXTERNAL_CACHE_REFRESH_COUNTER, TAG_OUTCOME, SUCCESS).increment())
.doOnError(sourceThrowable -> {
log.debug(LOG_REFRESH_CACHE + "Error while refreshing external cache {}", sourceThrowable.getMessage());
meterRegistry.counter(METRIC_EXTERNAL_CACHE_REFRESH_COUNTER, TAG_OUTCOME, FAILURE).increment();
}).subscribe();
}

此外，在上面的方法中，您可以看到对storeOffersInExternalCache的调用

public void storeOffersInExternalCache(OfferData offerData, Duration ttl) {
log.info(LOG_STORING_OFFER_DATA + "Storing the offer data in external cache...");
redisAdapter.storeOffers(offerData, ttl);
}

public void storeOffers(OfferData offerData, Duration ttl) {
Mono.fromRunnable(() -> redisClient.storeSerializedOffers(serializeFromDomain(offerData), ttl)
.doOnNext(status -> {
if (Boolean.TRUE.equals(status)) {
log.info(LOG_STORE_OFFERS + "Data stored in redis.");
meterRegistry.counter(METRIC_REDIS_STORE_OFFERS, TAG_OUTCOME, SUCCESS).increment();
} else {
log.error(LOG_STORE_OFFERS + "Unable to store data in redis.");
meterRegistry.counter(METRIC_REDIS_STORE_OFFERS, TAG_OUTCOME, FAILURE).increment();
}
}).retryWhen(Retry.backoff(redisRetryProperties.getMaxAttempts(), redisRetryProperties.getWaitDuration()).jitter(redisRetryProperties.getBackoffJitter()))
.doOnError(throwable -> {
meterRegistry.counter(METRIC_REDIS_STORE_OFFERS, TAG_OUTCOME, FAILURE).increment();
log.error(LOG_STORE_OFFERS + "Unable to store data in redis. Error: [{}]", throwable.getMessage());
})).subscribeOn(Schedulers.boundedElastic());
}

Redis客户端

@Slf4j
@Component
public class RedisClient {
private final ReactiveRedisTemplate<String, String> reactiveRedisTemplate;
private final ReactiveValueOperations<String, String> reactiveValueOps;
public RedisClient(@Qualifier("reactiveRedisTemplate") ReactiveRedisTemplate<String, String> reactiveRedisTemplate) {
this.reactiveRedisTemplate = reactiveRedisTemplate;
this.reactiveValueOps = reactiveRedisTemplate.opsForValue();
}
Mono<Optional<String>> fetchSerializedOffers() {
return reactiveValueOps.get(OFFER_DATA_KEY).map(Optional::ofNullable);
}
Mono<Boolean> storeSerializedOffers(String serializedOffers, Duration ttl) {
return reactiveValueOps.set(OFFER_DATA_KEY, serializedOffers, ttl);
}
Mono<Duration> getTtl() {
return reactiveRedisTemplate.getExpire(OFFER_DATA_KEY);
}
}

现在我担心的是：

1. 如果我不在这些Mono流上调用subscribe方法，这些方法甚至不会执行。这是公平的，因为在有人订阅之前，它们不会执行
2. 正如我正确理解的那样，subscribe是一个阻塞调用。这违背了反应式编程的全部目的。不是吗
3. 我寻找了几种方法来实现这一点，其中一种已经在上面展示了。我尝试调用Mono.fromRunnable中的一个方法，但这也不是一个很好的方法。(在StackOverflow中的另一个线程上阅读(

那么，我上面采用的方法是否不正确？我们如何执行没有人订阅的Mono流？

Answer 1

回答您的问题2(这似乎是您问题中唯一真正的疑问(。不是。block()(https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html#block--)是预订CCD_ 7或CCD_。另一方面subscribe()(https://projectreactor.io/docs/core/release/api/reactor/core/publisher/Mono.html#subscribe--)订阅CCD_ 10或CCD_。