暴露 Reactor 指标

Doubao 中英对照 Exposing Reactor metrics

目录

• 1. 调度器指标
• 2. Publisher 指标
  • 2.1. 标签（Tags）
  • 2.2. 观测（Observation）
• 3. Reactor-Core-Micrometer 模块的 Meter 与标签
  • 3.1. Micrometer.metrics()
  • 3.2. Micrometer.timedScheduler()
  • 3.3. Micrometer.observation()

Project Reactor 是一个为高性能和更好资源利用而设计的库。但要真正理解一个系统的性能，最好能够监控它的各个组件。

正因如此，Reactor 通过 reactor-core-micrometer 模块提供了与 Micrometer 的内置集成。该模块在 2022.0 BOM 版本中引入，它提供了对 Micrometer 的显式依赖，从而能够为指标和观测提供经过精细调校的 API。

备注

在 Reactor-Core 3.5.0 之前，指标是以操作符形式实现的：当 Micrometer 不在 classpath 上时，这些操作符为 no-op。

reactor-core-micrometer API 要求用户显式提供某种形式的_注册器_（registry），而不是依赖一个硬编码的全局注册器。当对具有原生命名或标签概念的类应用插桩时，这些 API 会尝试在响应式链中发现这类元素。否则，API 会要求在提供注册器的同时，还提供一个用于命名 meter 的_前缀_（prefix）。

1. 调度器指标

Reactor 中的每个异步操作都通过线程与调度器中所述的 Scheduler 抽象完成。因此，监控您的调度器、留意开始变得可疑的关键指标并据此做出反应，就显得十分重要。

reactor-core-micrometer 模块提供了一个「计时」Scheduler 包装器，它会对通过其提交的任务进行测量，用法如下：

Scheduler originalScheduler = Schedulers.newParallel("test", 4);

Scheduler schedulerWithMetrics = Micrometer.timedScheduler(
	originalScheduler, 1
	applicationDefinedMeterRegistry, 2
	"testingMetrics", 3
	Tags.of(Tag.of("additionalTag", "yes")) 4
);

• 待包装的 Scheduler

• 用于发布指标的 MeterRegistry

• 命名 meter 时使用的前缀。例如，这会创建一个名为 testingMetrics.scheduler.tasks.completed 的 meter。

• 可选标签，会添加到为该包装 Scheduler 创建的所有 meter 上

important

当包装一个通用 Scheduler（例如 Schedulers.single()）或一个在多处使用的 Scheduler 时，只有通过 Micrometer#timedScheduler 返回的包装实例所提交的 Runnable 任务才会被插桩。

关于产生的 meter 及关联的默认标签，请参阅 Micrometer.timedScheduler()。

2. Publisher 指标

有时，在响应式管道的某个阶段记录指标会很有用。

一种做法是：从提供给 tap 操作符的自定义 SignalListener 中，手动把值推送到您所选的指标后端。

实际上，reactor-core-micrometer 模块通过 Micrometer#metrics API 提供了开箱即用的实现。考虑下面这条管道：

listenToEvents()
    .doOnNext(event -> log.info("Received {}", event))
    .delayUntil(this::processEvent)
    .retry()
    .subscribe();

要为这个源 Flux（由 listenToEvents() 返回）启用指标，我们需要打开指标收集：

listenToEvents()
    .name("events") 1
    .tap(Micrometer.metrics( 2
        applicationDefinedMeterRegistry 3
    ))
    .doOnNext(event -> log.info("Received {}", event))
    .delayUntil(this::processEvent)
    .retry()
    .subscribe();

• 响应式管道此阶段的每个指标都会使用 "events" 作为命名前缀（可选，默认为 reactor 前缀）。

• 我们使用 tap 操作符，并结合 reactor-core-micrometer 中提供的 SignalListener 实现来收集指标。

• 与该模块中的其他 API 一样，需要显式提供用于发布指标的 MeterRegistry。

所暴露指标的详细信息见 Micrometer.metrics()。

2.1. 标签（Tags）

除了 Micrometer.metrics() 中所述的通用标签之外，用户还可以通过 tag 操作符为自己的响应式链添加自定义标签：

listenToEvents()
    .name("events") 1
    .tag("source", "kafka") 2
    .tap(Micrometer.metrics(applicationDefinedRegistry)) 3
    .doOnNext(event -> log.info("Received {}", event))
    .delayUntil(this::processEvent)
    .retry()
    .subscribe();

• 此阶段的每个指标都会以 "events" 前缀标识。

• 设置自定义标签 "source"，值为 "kafka"。

• 除通用标签外，所有上报的指标都会带上 source=kafka 标签。

请注意，根据您所使用的监控系统，在使用标签时，使用名称可能被视为必需，否则两条默认命名的序列之间会产生不同的标签集合。像 Prometheus 这样的系统还可能要求同名的每个指标都具有完全相同的标签集合。

2.2. 观测（Observation）

除了完整的指标外，reactor-core-micrometer 模块还提供了一种基于 Micrometer Observation 的替代方案。根据配置和运行时 classpath，一个 Observation 可以转化为计时器（timer）、跨度（span）、日志语句或任意组合。

响应式链可以通过 tap 操作符和 Micrometer.observation 工具来观测，如下：

listenToEvents()
    .name("events") 1
    .doOnNext(event -> log.info("Received {}", event))
    .delayUntil(this::processEvent)
    .tap(Micrometer.observation( 2
		applicationDefinedRegistry)) 3
    .retry()
    .subscribe();

• 该管道的 Observation 会以 "events" 前缀标识。

• 我们把 tap 操作符与 observation 工具结合使用，向上游操作符提供一个匿名 observation。

• 必须提供一个注册器，用于发布观测结果。请注意这是一个 ObservationRegistry。

观测及其标签的详细信息见 Micrometer.observation()。

注意

用 Micrometer.observation() 收集的 meter 集合，与 Micrometer.metrics() 提供的 meter 集合不同。Micrometer.observation() 覆盖整条响应式链的生命周期，而 Micrometer.metrics() 提供链内单个信号的一些洞察。由于 Micrometer.observation() 的通用性（例如除指标外还提供追踪），它可能需要包装工作单元（如 Mono.just(processedItem)），并用类似 flatMap 的操作符观测一条子链。

您还可以通过 Micrometer.observation(ObservationRegistry registry, Function<ObservationRegistry, Observation> observationSupplier)，用自己的 Observation 供应器来完全定制 Micrometer 的 Observation，如下：

listenToEvents()
    .name("events") 1
    .doOnNext(event -> log.info("Received {}", event))
    .delayUntil(this::processEvent)
    .tap(Micrometer.observation( 2
    	applicationDefinedRegistry, 3
    	registry -> Observation.createNotStarted( 4
    		myConvention, 5
            myContextSupplier, 6
            registry)))
    .retry()
    .subscribe();

• 该管道的 Observation 会以 "events" 前缀标识。

• 我们把 tap 操作符与 observation 工具结合使用，向上游操作符提供一个自定义 observation。

• 必须提供一个注册器，用于发布观测结果。请注意这是一个 ObservationRegistry。

• 我们提供自己的函数来创建 Observation。

• 该函数带一个自定义 ObservationConvention。

• 以及一个自定义 Supplier<Context>。

3. Reactor-Core-Micrometer 模块的 Meter 与标签

3.1. Micrometer.metrics()

下面是 metrics tap listener 特性所使用的 meter 列表，通过 Micrometer.metrics(MeterRegistry meterRegistry) 暴露。

important

请注意，下面的指标使用动态的 %s 前缀。当应用到使用了 name(String n) 操作符的 Flux 或 Mono 上时，它会被替换为 n。否则，会被替换为默认值 "reactor"。

3.1.1. 流持续时间

测量从订阅到序列终止或取消之间经过的时长。会添加一个 TerminationTags#STATUS 标签，用以指明是哪个事件结束了该计时器（"completed"、"completedEmpty"、"error" 或 "cancelled"）。

指标名 %s.flow.duration——由于包含 %s，该名称是动态的，会在运行时解析。类型 distribution summary（分布摘要）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

表 1. 低基数键（Low cardinality Keys）

名称	说明
exception (required)	当 STATUS 为 "error" 时，FLOW_DURATION 用此标签记录发生的异常。
status (required)	终止状态： - "completed"：序列在有 onNext 的情况下以 onComplete 终止 - "completedEmpty"：序列在 onComplete 之前没有任何 onNext 就终止 - "error"：序列以 onError 终止 - "cancelled"：序列取消了其订阅
type (required)	序列的类型（"Flux" 或 "Mono"）。

3.1.2. 畸形源事件

记录从畸形源收到的事件数量（例如在 onComplete 之后又出现 onNext）。

指标名 %s.malformed.source——由于包含 %s，该名称是动态的，会在运行时解析。类型 counter（计数器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

表 2. 低基数键（Low cardinality Keys）

名称	说明
type (required)	序列的类型（"Flux" 或 "Mono"）。

3.1.3. On Next 延迟

测量相邻两个 onNext 之间（或第一个 onNext 与 onSubscribe 事件之间）的延迟。

指标名 %s.onNext.delay——由于包含 %s，该名称是动态的，会在运行时解析。类型 timer（计时器），基准单位 nanoseconds（纳秒）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

important

Micrometer 内部以 nanoseconds 作为基准单位。然而，实际基准单位由各后端决定（例如 Prometheus 使用秒）。

表 3. 低基数键（Low cardinality Keys）

名称	说明
type (required)	序列的类型（"Flux" 或 "Mono"）。

3.1.4. 请求量

记录所有订阅者向一个具名序列（例如 Flux.name(String)）请求的总量，直到至少有一个请求了无界数量为止。

指标名 %s.requested——由于包含 %s，该名称是动态的，会在运行时解析。类型 distribution summary（分布摘要）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

表 4. 低基数键（Low cardinality Keys）

名称	说明
type (required)	序列的类型（"Flux" 或 "Mono"）。

3.1.5. 订阅数

记录对一个序列的订阅次数。

指标名 %s.subscribed——由于包含 %s，该名称是动态的，会在运行时解析。类型 counter（计数器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

表 5. 低基数键（Low cardinality Keys）

名称	说明
type (required)	序列的类型（"Flux" 或 "Mono"）。

3.2. Micrometer.timedScheduler()

下面是 TimedScheduler 特性所使用的 meter 列表，通过 Micrometer.timedScheduler(Scheduler original, MeterRegistry meterRegistry, String metricsPrefix) 暴露。

important

请注意，下面的指标使用动态的 %s 前缀。实际中它会被替换为所提供的 metricsPrefix。

3.2.1. 活跃任务

一个 LongTaskTimer，反映当前正在运行的任务。请注意，它反映所有类型的活跃任务，包括延迟调度或周期性调度的任务（每一次迭代都被视为一个活跃任务）。

指标名 %s.scheduler.tasks.active——由于包含 %s，该名称是动态的，会在运行时解析。类型 long task timer（长任务计时器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

important

Micrometer 内部以 nanoseconds 作为基准单位。然而，实际基准单位由各后端决定（例如 Prometheus 使用秒）。

3.2.2. 已完成任务

一个计时器，反映已执行完毕的任务。请注意，它反映所有类型的活跃任务，包括带延迟或周期性的任务（每一次迭代都被视为一个单独的已完成任务）。

指标名 %s.scheduler.tasks.completed——由于包含 %s，该名称是动态的，会在运行时解析。类型 timer（计时器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

important

Micrometer 内部以 nanoseconds 作为基准单位。然而，实际基准单位由各后端决定（例如 Prometheus 使用秒）。

3.2.3. 待处理任务

一个 LongTaskTimer，反映那些被提交以立即执行、但因调度器已达最大容量而无法立即开始的任务。请注意，只统计通过 Scheduler#schedule(Runnable) 和 Scheduler.Worker#schedule(Runnable) 的即时提交。

指标名 %s.scheduler.tasks.pending——由于包含 %s，该名称是动态的，会在运行时解析。类型 long task timer（长任务计时器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

important

Micrometer 内部以 nanoseconds 作为基准单位。然而，实际基准单位由各后端决定（例如 Prometheus 使用秒）。

3.2.4. 已提交任务

一个计数器，每当提交一个任务时加一（通过 Scheduler 与 Scheduler.Worker 上的任一 schedule 方法）。

请注意，实际上有 4 个计数器，可通过 SubmittedTags#SUBMISSION 标签区分。因此可以把所有这些计数器之和与 TASKS_COMPLETED 计数器进行比较。

指标名 %s.scheduler.tasks.submitted——由于包含 %s，该名称是动态的，会在运行时解析。类型 counter（计数器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

表 6. 低基数键（Low cardinality Keys）

名称	说明
submission.type (required)	提交类型： - "direct"：对应 Scheduler#schedule(Runnable) - "delayed"：对应 Scheduler#schedule(Runnable,long,TimeUnit) - "periodic_initial"：对应初始延迟之后的 Scheduler#schedulePeriodically(Runnable,long,long,TimeUnit) - "periodic_iteration"：对应 Scheduler#schedulePeriodically(Runnable,long,long,TimeUnit) 后续的周期性迭代

3.3. Micrometer.observation()

下面是 observation tap listener 特性所使用的 meter 列表，通过 Micrometer.observation(ObservationRegistry registry) 暴露。

这是匿名观测（ANONYMOUS observation），但您可以使用 name(String) 操作符，创建一个带自定义名称的类似 Observation。

备注

您还可以通过 Micrometer.observation(ObservationRegistry registry, Function<ObservationRegistry, Observation> observationSupplier)，用自己的 Observation 供应器完全定制 Micrometer 的 Observation，从而配置它的各项属性（名称、上下文名称、低基数与高基数键等）。

3.3.1. 匿名

Micrometer.observation() 的匿名版本，当序列尚未通过例如 Flux#name(String) 操作符显式命名时使用。

指标名 reactor.observation。类型 timer（计时器）。

指标名 reactor.observation.active。类型 long task timer（长任务计时器）。

important

在启动 Observation 之后添加的 KeyValues，可能不会出现在 *.active 指标中。

important

Micrometer 内部以 nanoseconds 作为基准单位。然而，实际基准单位由各后端决定（例如 Prometheus 使用秒）。

表 7. 低基数键（Low cardinality Keys）

名称	说明
reactor.status (required)	序列的状态，表明它如何终止（"completed"、"completedEmpty"、"error" 或 "cancelled"）。
reactor.type (required)	序列的类型，即 "Flux" 或 "Mono"。