使用elapsed()函数在Mono上工作？

Question

在反应式编程中，我正在尝试获取读取redis的执行时间，在查找文档时，我能够看到elapsed()方法将执行相同的操作，并实现如下代码。

Flux.fromIterable(getActions(httpHeaders))
                .parallel()
                .runOn(Schedulers.parallel())
                .flatMap(actionFact -> methodToReadFromCache(actionFact))
                .sequential();

public Mono<ActionFact> methodToReadFromCache(actionFact) {
    return Mono.fromCallable(() -> getKey(actionFact))
                .flatMap(cacheKey ->
                  redisOperations.hasKey(key)
                                .flatMap(aBoolean -> {
                                    if (aBoolean) {
                                        return redisOperations.opsForValue().get(cacheKey);
                                    }
                                    return authzService.getRolePermissions(actionFact)
                                            .flatMap(policySetResponse ->
                                                    //save in cache
                                            );
                                })
                                .elapsed()
                                .flatMap(lambda -> {
                                    LOG.info("cache/service processing key:{}, time:{}", key, lambda.getT1());
                                    return Mono.just(lambda.getT2());
                                });

输出：

cache/service processing key:KEY1, time:3 
cache/service processing key:KEY2, time:4 
cache/service processing key:KEY3, time:18 
cache/service processing key:KEY4, time:34 
cache/service processing key:KEY5, time:46 
cache/service processing key:KEY6, time:57 
cache/service processing key:KEY7, time:70 
cache/service processing key:KEY8, time:81 
cache/service processing key:KEY9, time:91 
cache/service processing key:KEY10, time:103
cache/service processing key:KEY11, time:112
cache/service processing key:KEY12, time:121
cache/service processing key:KEY13, time:134
cache/service processing key:KEY14, time:146
cache/service processing key:KEY15, time:159

我期望每个缓存请求花费的时间将小于5毫秒，就像第一个和第二个请求一样，但情况并非如此。elapsed()是否将当前获取时间添加到累加器中？根据我的理解，从flux发出的每个项目都是独立的？

Answer 1

Mono#elapsed测量从订阅Mono到Mono发出项目的时刻(onNext)之间的时间。

在您的示例中，导致订阅和计时器启动的是调用methodToReadFromCache的外部并行化flatMap。

导致onNext并因此计时的是hasKey和if/else部分的组合(redisOperations.opsForValue().get(cacheKey) vs authzService)。

外部flatMap至少应该和CPU一样多，因为我们处于并行模式。

但计时不对称的事实可能暗示了这样一个事实，即某些东西要么是阻塞的，要么是容量有限的。例如，redisTemplate一次只能处理几个密钥吗？

Answer 2

根据文档

我想将排放与timing (Tuple2<Long, T>)

…关联起来​

订阅后的

• ：elapsed

从时间的黎明(好吧，是计算机时间)开始的

• ：timestamp

elapsed是自订阅以来测量的时间。所以你订阅了，它开始发射，时间会随着你订阅你的服务的时间延长而增加。

official docs