如何解决 Mono.fromFuture 方法调用不一致的问题？

Question

我正在开发一个运行 Webflux（版本 5.3）的 Spring Boot 项目。

在 REST 端点中，两个不同的 AWS DynamoDB 账户有两个异步更新，并通过 Mono.when() 聚合在一起。在调用 Mono.when (metric-1) 之前报告了一个 Datadog 指标，并在 DynamoDB 更新方法中报告了成功 (metric-2) 和失败 (metric-3) 的指标。

Mono<Void> result = Mono.empty();
        if (//valid check) {
            //Report Datadog metric-1
            result = Mono.when(updateDynamoDBTable(event, dynamoDbAsyncClient1),
                               updateDynamoDBTable(event, dynamoDbAsyncClient2));
        }
        return result;

//call the given client to update the given record
private updateDynamoDBTable(Event event, DynamoDbAsyncClient dynamoDbAsyncClient){
    return Mono.defer(() -> Mono.just(event))
                .map(this::toDynamoRequest)
                .flatMap(item -> Mono.fromFuture(dynamoDbAsyncClient.updateItem(toDynamoRequest)))
                .doOnNext(uir -> //Report Datadog metric 2)
                .onErrorResume(e -> {
                      //Report Datadog metric 3
                        };
                    return Mono.empty();
                });
}  

当此应用程序在生产环境中高负载下运行时，我发现指标计数不匹配。 Mono.when 的调用次数远高于 DynamoDB 成功指标。 Datadog 没有记录或报告任何故障。

如何调试这种不匹配？当没有定义默认线程池时，WebFlux项目使用哪个线程池？

Answer 1

我添加了一个

.doFinally

方法来跟踪信号类型并报告取消的指标。我现在可以看到丢失的电话已被取消。

这是更新后的代码：

    Mono<Void> result = Mono.empty();
            if (//valid check) {
                //Report Datadog metric-1
                result = Mono.when(updateDynamoDBTable(event, dynamoDbAsyncClient1),
                                   updateDynamoDBTable(event, dynamoDbAsyncClient2));
            }
            return result;
    
    //call the given client to update the given record
    private updateDynamoDBTable(Event event, DynamoDbAsyncClient dynamoDbAsyncClient){
        return Mono.defer(() -> Mono.just(event))
                    .map(this::toDynamoRequest)
                    .flatMap(item -> Mono.fromFuture(dynamoDbAsyncClient.updateItem(toDynamoRequest)))
                    .doOnNext(uir -> //Report Datadog metric 2)
                    .onErrorResume(e -> {
                          //Report Datadog metric 3
                            };
                        return Mono.empty();
                    })
                    .doFinally(signalType -> {
                          if (signalType == Cancel) {
                          //Report Datadog metric 4
                            };
                        return Mono.empty();
                    });
    }  

我现在可以看到调用数量现在匹配 - Mono.when (metric-1) = success (metric-2) + failure (metric-3) + Canceled (metric-4)。

我仍在努力找出取消的原因。