执行时间响应式编程

Question

这是getFavouriteDetails()在反应式编程中找到方法 ( )执行时间的理想方法吗？

public List<Favourites> getFavouriteDetails(String userId){
    userService.getFavorites(userId) 
               .flatMap(favoriteService::getDetails) 
               .switchIfEmpty(suggestionService.getSuggestions()) 
               .take(5) 
               .publishOn(UiUtils.uiThreadScheduler()) 
               .subscribe(uiList::show, UiUtils::errorPopup)
               .flatMap(a -> Mono.subscriberContext().map(ctx -> {
                         log.info("Time taken : " + Duration.between(ctx.get(key), Instant.now()).toMillis() + " milliseconds.");
                         return a;
                     }))
               .subscriberContext(ctx -> ctx.put(key, Instant.now()))
}

Answer 1

对方法进行计时，Java 中最基本的方法是使用long System.nanoTime(). Instant和System.currentTimeMillis用于挂钟操作，不能保证单调或足够精确......

在 Reactor 中，为了测量一个序列完成所需的时间，您通常需要在订阅上开始计时（在您订阅之前不会发生任何事情）并在 a 内停止计时doFinally（只要它在主序列一侧执行一些代码）完成、错误或被取消）。

然而，在这里您是自己订阅，因此不存在多次订阅的风险。因此，您可以取消“订阅开始时间”约束。

它给了我们这样的东西：

public List<Favourites> getFavouriteDetails(String userId){
    final long start = System.nanoTime();
    userService.getFavorites(userId) 
               .flatMap(favoriteService::getDetails) 
               .switchIfEmpty(suggestionService.getSuggestions()) 
               .take(5) 
               .publishOn(UiUtils.uiThreadScheduler())
               .doFinally(endType -> log.info("Time taken : " + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) + " milliseconds."))
               .subscribe(uiList::show, UiUtils::errorPopup);
    //return needed!
}

请注意，还有一个elapsed()运算符，它测量订阅和第一个 onNext 之间的时间，以及后续 onNext 之间的时间。它输出 a Flux<Tuple2<Long, T>>，您可以聚合 long 以获得总体时序，但T在这种情况下，这会失去 s 的“实时”性质。

Answer 2

两种方法可确保您仅在订阅时测量执行时间 -

• 使用 将 Mono 包裹在 Flux 周围flatMapMany。这也返回一个 Flux 。
• 使用 AtomicReference，在 中设置时间onSubscribe并记录经过的时间doFinally。

示例代码 -

timeFluxV1(getFavouriteDetails(userId)).subscribe(uiList::show, UiUtils::errorPopup);

timeFluxV1(getFavouriteDetails(userId)).subscribe(uiList::show, UiUtils::errorPopup);
 
private <T>  Flux<T> timeFluxV1(Flux<T> flux) {
    return Mono.fromSupplier(System::nanoTime)
             .flatMapMany(time -> flux.doFinally(sig -> log.info("Time taken : " + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - time) + " milliseconds.")));
}


private <T>  Flux<T> timeFluxV2(Flux<T> flux) {
    AtomicReference<Long> startTime = new AtomicReference<>();
    return flux.doOnSubscribe(x -> startTime.set(System.nanoTime()))
            .doFinally(x -> log.info("Time taken : " + TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime.get()) + " milliseconds."));
}

public Flux<Favourites> getFavouriteDetails(String userId) {
    return userService.getFavorites(userId)
               .flatMap(favoriteService::getDetails) 
               .switchIfEmpty(suggestionService.getSuggestions()) 
               .take(5) 
               .publishOn(UiUtils.uiThreadScheduler());
}