如何测量 webflux WebClient 方法的执行时间？

Question

我准备了一堆要并行发送到外部 Web 服务的请求。在此流程中，我继续直接处理响应（例如，将某些内容插入数据库）。

问题：我想跟踪最大请求时间（针对一个请求！），不包括处理。但正如所写，这只会跟踪全局时间，包括任何子进程：

StopWatch watch = new StopWatch();
watch.start();

Flux.fromIterable(requests)
    .flatMap(req -> webClient.send(req, MyResponse.class)
            .doOnSuccess(rsp -> processResponse(rsp))) //assume some longer routine
    .collectList()
    .block();
    
watch.stop();
System.out.println(w.getTotalTimeMillis());
            

问题：如何测量请求花费的最长时间（不包括processResponse()时间）？

Answer 1

当在单声道上使用 elapsed 时，您将返回元组的单声道，其中包含经过的时间和原始对象。您必须打开它们才能使用它们。我在测试中编写了一个示例（对您的代码进行了一些简化）以查看它的工作原理：

@Test
public void elapsed() {

    Flux.fromIterable(List.of(1, 2, 3, 4, 5))
        .flatMap(req -> Mono.delay(Duration.ofMillis(100L * req))
                            .map(it -> "response_" + req)
                            .elapsed()
                            .doOnNext(it -> System.out.println("I took " + it.getT1() + " MS"))
                            .map(Tuple2::getT2)
                            .doOnSuccess(rsp -> processResponse(rsp)))
        .collectList()
        .block();

}

@SneakyThrows
public void processResponse(Object it) {
    System.out.println("This is the response: " + it);
    Thread.sleep(1000);
}

输出如下所示：

I took 112 MS
This is the response: response_1
I took 205 MS
This is the response: response_2
I took 305 MS
This is the response: response_3
I took 403 MS
This is the response: response_4
I took 504 MS
This is the response: response_5

这些数字代表延迟（在您的情况下是 webClient.send()）和反应式管道本身的一点开销。它是在订阅（当特定请求的 flatMap 运行时发生）和下一个信号（在我的例子中来自地图的结果，在你的例子中是 webclient 请求的结果）之间计算的

你的代码看起来像这样：

Flux.fromIterable(requests)
        .flatMap(req -> webClient.send(req, MyResponse.class)
                                 .elapsed()
                                 .doOnNext(it -> System.out.println("I took " + it.getT1() + " MS"))
                                 .map(Tuple2::getT2)
                                 .doOnSuccess(rsp -> processResponse(rsp))) //assume some longer routine
        .collectList()
        .block();

请注意，如果您想使用秒表代替，也可以通过执行以下操作来实现：

Flux.fromIterable(List.of(1, 2, 3, 4, 5)).flatMap(req -> {
            StopWatch stopWatch = new StopWatch();
            return Mono.fromRunnable(stopWatch::start)
                       .then(Mono.delay(Duration.ofMillis(100L * req)).map(it -> "response_" + req).doOnNext(it -> {
                           stopWatch.stop();
                           System.out.println("I took " + stopWatch.getTime() + " MS");
                       }).doOnSuccess(this::processResponse));
        }).collectList().block();

但我个人会推荐 .elapsed() 解决方案，因为它更干净一些。