网络流量订户

Question

我目前面临一个关于在switchIfEmpty函数中保存redis的问题。这可能与我在反应性编程方面很新有关，不幸的是，我很难找到适当的例子。

然而，我已经能够解决它，但我很确定有更好的方法来做它。我现在写的是：

public Mono<ResponseEntity<BaseResponse<Content>>> getContent(final String contentId, final String contentType){
    return redisRepository.findByKeyAndId(REDIS_KEY_CONTENT, contentId.toString()).cast(Content.class)
               .map(contentDTO -> ResponseEntity.status(HttpStatus.OK.value())
                                                .body(new BaseResponse<>(HttpStatus.OK.value(), HttpStatus.OK.getReasonPhrase(), contentDTO)))
               //here I have to defer, otherwise It would never wait for the findByKeyAndId 
               .switchIfEmpty(Mono.defer(() -> {
                   Mono<ResponseEntity<BaseResponse<Content>>> responseMono = contentService.getContentByIdAndType(contentId, contentType);
                   
                   //so far I understood I need to consume every stream I have, in order to actually carry out the task otherwise will be there waiting for a consumer.
                   //once I get what I need from the cmsService I need to put it in cache and till now this is the only way I've been able to do it
                   responseMono.filter(response -> response.getStatusCodeValue() == HttpStatus.OK.value())
                               .flatMap(contentResponse -> redisRepository.save(REDIS_KEY_CONTENT, contentId.toString(), contentResponse.getBody().getData()))
                                        .subscribe();
                   //then I return the Object I firstly retrived thru the cmsService
                   return responseMono;
               }
    ));
}

有没有更好的方法的线索或建议？提前感谢您的帮助！

Answer 1

从最佳实践的角度来看，有几件事情并不是很好，而且这可能并没有做你认为它正在做的事情：

• --你在订阅自己，这通常是一个明确的标志除特殊情况外，订阅一般应留待框架处理。这也是你需要Mono.defer()的原因--通常，除非框架在正确的时间订阅了发布服务器，否则什么都不会发生，而你自己正在管理该出版商的订阅生命周期。
• 框架仍然会在那里订阅您的内部发布服务器，只是返回的Mono不会对它的结果做任何事情。因此，您可能要调用contentService.getContentByIdAndType()两次，而不仅仅是一次--一次是在订阅时，另一次是在内部发布服务器上的框架later.
• Unrelated创建一个“失火和遗忘”类型模型时，这意味着当您的反应性方法返回时，您不知道redis是否已经保存了它，这可能会导致问题，如果您随后依赖于上面的结果later.
• Unrelated，但是contentId已经是一个字符串，您不需要在它上调用toString() :-)

相反，您可以考虑在您的delayUntil块中使用switchIfEmpty() --这将允许您在响应代码正常的情况下将值保存到redis，延迟到发生这种情况，并在完成时保留原始值。代码可能如下所示(如果没有完整的示例，很难判断这是否完全正确，但它应该给您一个想法)：

return redisRepository
        .findByKeyAndId(REDIS_KEY_CONTENT, contentId).cast(Content.class)
        .map(contentDTO -> ResponseEntity.status(HttpStatus.OK.value()).body(new BaseResponse<>(HttpStatus.OK.value(), HttpStatus.OK.getReasonPhrase(), contentDTO)))
        .switchIfEmpty(
                contentService.getContentByIdAndType(contentId, contentType)
                        .delayUntil(response -> response.getStatusCodeValue() == HttpStatus.OK.value() ?
                                redisRepository.save(REDIS_KEY_CONTENT, contentId, contentResponse.getBody().getData()) :
                                Mono.empty())
        );