Micrometer 简介
Micrometer 是 JVM 应用的指标门面（Facade），类似 SLF4J 之于日志。它支持多种后端：Prometheus、Graphite、Datadog、InfluxDB 等。
eventd v2 使用 Prometheus 作为指标存储和查询引擎。
依赖
<dependency>
<groupId>io.micrometer</groupId>
<artifactId>micrometer-registry-prometheus</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
三大基础指标类型
Counter：只增不减的计数器
import io.micrometer.core.instrument.MeterRegistry;
@Service
public class OrderService {
private final MeterRegistry registry;
public OrderService(MeterRegistry registry) {
this.registry = registry;
}
public Mono<Order> create(Order order) {
return repo.save(order)
.doOnNext(saved -> registry.counter("orders.created",
"service", "eventd-api", "status", "SUCCESS").increment())
.doOnError(err -> registry.counter("orders.created",
"service", "eventd-api", "status", "ERROR").increment());
}
}
Prometheus 输出：
orders_created_total{service="eventd-api",status="SUCCESS"} 42
orders_created_total{service="eventd-api",status="ERROR"} 1
同一个指标名下的标签键最好保持一致：缺 status 的序列在 Prometheus 里是另一条时间序列，查询和聚合时必须处理缺失标签，否则 sum by (status) 会漏掉它。
Timer：计时 + 统计分桶
private final MeterRegistry registry;
private final Timer orderTimer;
public OrderService(MeterRegistry registry) {
this.registry = registry;
this.orderTimer = Timer.builder("orders.create.latency")
.description("Order creation latency")
.serviceLevelObjectives(
Duration.ofMillis(50), Duration.ofMillis(100), Duration.ofMillis(500))
.publishPercentileHistogram()
.register(registry);
}
public Mono<Order> create(Order order) {
return Mono.defer(() -> {
Timer.Sample sample = Timer.start(registry);
return repo.save(order)
.doFinally(sig -> sample.stop(orderTimer)); // 链结束（成功/失败/取消）时记录
});
}
Timer.record(Supplier) 只是同步地调用一次 supplier：orderTimer.record(() -> repo.save(order)) 计到的是「组装 Mono」的耗时，几乎为零，数据库调用根本没有被包住。响应式链必须用 Timer.Sample + doFinally，或交给 Observation API。Timer 也没有接受 Publisher 的重载，所以 transformDeferred(orderTimer::record) 编译不过。
Gauge：瞬时值
Gauge.builder("orders.pending.count", orderRepo, repo -> {
// ❌ 错误：Flux.block() 阻塞了！
return repo.countByStatus("PENDING").block();
})
.tag("status", "pending")
.register(registry);
上面的 block() 在 Gauge 的回调中阻塞线程，错误。改用已缓存的值：
AtomicLong pendingCount = new AtomicLong(0);
// 定时更新 Gauge 值（非阻塞）
Flux.interval(Duration.ofSeconds(10))
.flatMap(i -> orderRepo.countByStatus("PENDING"))
.subscribe(pendingCount::set);
Gauge.builder("orders.pending.count", pendingCount, AtomicLong::get)
.tag("status", "pending")
.register(registry);
Gauge 回调必须是非阻塞的、返回值立即可得的。上面的 Flux.interval 会一直跑，应用关闭要调用返回的 Disposable.dispose()，或者把更新逻辑放进 @Scheduled，避免测试和应用重启时残留后台定时任务。
另一个容易忽略的点：Micrometer 的 Gauge 默认以弱引用持有被观测对象，如果 AtomicLong 只被一个局部变量引用，随时可能被 GC 回收，之后读到的就是 NaN。这里订阅回调捕获了 pendingCount，所以暂时是安全的；更稳妥的做法是把它作为 OrderService 的字段长期持有，或在 builder 上调用 .strongReference(true)。
Reactor Metrics Hook
Reactor 内置 metrics() 操作符，可自动收集流级指标：
Flux.interval(Duration.ofMillis(100))
.name("ticks") // 指标前缀
.metrics() // 启用指标收集
.map(i -> process(i))
.subscribe();
Reactor 3.8 为这条流注册的 Micrometer meter 是（.name("ticks") 决定前缀）：
ticks.subscribed Counter 订阅次数，tag: type=Flux
ticks.flow.duration Timer 订阅到终止的耗时，tag: type、status、exception
ticks.onNext.delay Timer 相邻 onNext 的间隔
ticks.requested DistributionSummary request(n) 的请求量分布
ticks.malformed.source 不在默认列表里：它只在 Reactor 检测到上游违反 Reactive Streams 规范（malformed source signal）时才按需注册，正常流不会产生这条序列。排查时不要把它当成「每个 .metrics() 流都该有」的指标。
这组 meter 默认注册到 Micrometer 的全局 registry（io.micrometer.core.instrument.Metrics.globalRegistry）。Spring Boot 的 MeterRegistryPostProcessor 在 management.metrics.use-global-registry=true（默认值）时，会把容器里的 MeterRegistry Bean 加进这个全局 registry，所以在 Boot 应用里通常不需要额外绑定：注入 MeterRegistry 后直接 .metrics()，/actuator/prometheus 就能看到上面这些序列。
只有在非 Boot 环境，或显式关闭了 use-global-registry 时，才需要自己把 registry 挂到全局：
io.micrometer.core.instrument.Metrics.addRegistry(registry);
Reactor 自带的 reactor.util.Metrics.MicrometerConfiguration.useRegistry(registry) 在 3.8 中仍可用，但已标记 @Deprecated，新代码不要再依赖它。如果不想依赖全局状态，更稳的做法是用 Micrometer 的 Observation（或 Spring Boot 对 WebClient/WebFlux 的自动观测）在调用边界埋点，而不是给每条内部流都挂 metrics()。
Prometheus 导出时点号变下划线，Timer 追加 _seconds，所以会看到 ticks_subscribed_total{type="Flux"}、ticks_flow_duration_seconds_count{type="Flux",status="completed"} 这类序列；status 的取值是 completed、completedEmpty、error、cancelled。
metrics() 只观测它上游的信号；写成 .map(...).metrics() 才能把 map 的耗时算进 flow.duration。另外它会为每条流注册多个 meter，开销与流的数量、事件频率相关。是否值得开要压测：给关键流开启，不要无差别地加在所有内部流上。
自定义标签与上下文
标签（tag/label）让指标可切片分析。在响应式系统中，标签值需要从 Reactor Context 中提取：
public Mono<Order> create(ServerWebExchange exchange, Order order) {
String requestId = exchange.getAttribute("requestId");
return repo.save(order)
.doOnNext(saved -> registry.counter("orders.created",
"status", saved.status(),
"symbol", saved.symbol()
).increment());
}
注意：不要在高基数维度上使用唯一值标签（如 userId、requestId）。Prometheus 每个唯一标签组合是一个时间序列，高基数会撑爆内存。
Prometheus 集成
Spring Boot Actuator 自动暴露 /actuator/prometheus 端点：
management:
endpoints:
web:
exposure:
include: prometheus,health,info
metrics:
tags:
application: eventd-v2
environment: production
Prometheus 配置抓取：
# prometheus.yml
scrape_configs:
- job_name: 'eventd'
static_configs:
- targets: ['eventd-api:8080']
metrics_path: '/actuator/prometheus'
scrape_interval: 15s
SLA 与告警规则
# rules.yml
groups:
- name: eventd-alerts
rules:
- alert: HighOrderLatency
expr: histogram_quantile(0.99, rate(orders_create_latency_seconds_bucket[5m])) > 0.5
for: 2m
annotations:
summary: "Order creation P99 latency > 500ms"
- alert: OrderErrors
expr: sum(rate(orders_created_total{status="ERROR"}[5m])) > 1
for: 1m
annotations:
summary: "Order creation error rate > 1/s"
小结
| 指标类型 | 用途 | API |
|---|---|---|
| Counter | 事件计数 | Counter.builder().register().increment() |
| Timer | 延迟 + 分桶 | Timer.start() + doFinally(sample.stop(timer)) |
| Gauge | 瞬时值 | Gauge.builder().register()，回调非阻塞 |
| Reactor Hook | 流级指标 | .name().metrics() |
下一节 10.2 分布式追踪 ：Brave/OpenTelemetry、ReactorContext 传播、Baggage 与 Span。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。