Mono流水线多阶段能否发布多指标？如何实现细粒度时长统计

你遇到的问题核心是：当前代码里的阶段metrics()统计的是整个父流的订阅生命周期，而非单个阶段自身的执行时长。因为flatMap中直接返回的processDataStage1(data)会和上游流共享订阅上下文，导致每个阶段的metrics()都从父流订阅开始计时，直到整个流结束，所以所有指标都显示总时长。

要实现各阶段的独立时长统计，最直接的方案是用Mono.defer()包装每个阶段的Mono，确保每个阶段拥有独立的订阅生命周期：

fetchSomeData()
  .flatMap(data -> Mono.defer(() -> processDataStage1(data))
     .name("processDataStage1")
     .metrics()
  )
  .flatMap(data -> Mono.defer(() -> processDataStage2(data))
     .name("processDataStage2")
     .metrics()
  )
  .name("total")
  .metrics()

为什么defer()能解决问题？ #

defer()的作用是延迟Mono的创建，直到订阅发生时才实例化新的Mono对象。在flatMap中使用它后：

• 当上游流产生数据触发flatMap时，才会订阅defer()内部的processDataStage1(data)
• 此时processDataStage1的metrics()会从这个阶段的订阅时刻开始计时，到该阶段完成/终止结束，得到的就是这个阶段的真实耗时
• 同理，processDataStage2也会拥有独立的订阅和计时周期

其他可选方案 #

如果需要复用阶段统计逻辑，可以用transformDeferred()替代直接的defer()，效果一致：

private <T> Function<Mono<T>, Mono<T>> stageMetrics(String stageName) {
    return mono -> Mono.defer(() -> mono)
                       .name(stageName)
                       .metrics();
}

// 使用方式
fetchSomeData()
  .flatMap(data -> processDataStage1(data).transform(stageMetrics("processDataStage1")))
  .flatMap(data -> processDataStage2(data).transform(stageMetrics("processDataStage2")))
  .name("total")
  .metrics();

这样修改后，processDataStage1.flow.duration、processDataStage2.flow.duration就会分别统计各自阶段的耗时，而total.flow.duration依然统计整个流水线的总时长，完全符合你的需求。

内容的提问来源于stack exchange，提问作者Marcel Riera