如何使用reactor-test的StepVerifier验证响应式流的metrics发布

StepVerifier本身不会阻断metrics上报逻辑，未触发上报通常是没有绑定测试专属的指标捕获注册表，或是未等待流的终止信号（完成/错误/取消）就提前做指标断言。可以按照以下方案实现：

• 测试初始化阶段创建SimpleMeterRegistry实例，作为测试过程中metrics的捕获容器
• 在待测试的响应式流上调用.metrics()操作符时，指定上报目标为刚才创建的测试注册表
• 使用StepVerifier正常订阅流，触发对应业务场景的信号，调用verify*系列方法等待流运行结束
• 从测试注册表中读取对应metrics项，直接做值断言即可

代码示例 #

import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.DistributionSummary;
import io.micrometer.core.instrument.simple.SimpleMeterRegistry;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;
import static org.assertj.core.api.Assertions.assertThat;

@Test
void testFluxMetrics() {
    // 初始化测试用指标注册表
    SimpleMeterRegistry testRegistry = new SimpleMeterRegistry();
    String metricPrefix = "custom.business.flux";

    // 构造待测试流，绑定测试注册表
    Flux<Integer> businessFlux = Flux.range(1, 5)
            .metrics(() -> testRegistry, metricPrefix);

    // StepVerifier正常触发流运行
    StepVerifier.create(businessFlux)
            .expectNext(1,2,3,4,5)
            .verifyComplete();

    // 断言请求数指标
    Counter requestCounter = testRegistry.find(metricPrefix + ".requests")
            .counter();
    assertThat(requestCounter.count()).isEqualTo(1);

    // 断言元素发出数量指标
    DistributionSummary onNextSummary = testRegistry.find(metricPrefix + ".onNext")
            .summary();
    assertThat(onNextSummary.count()).isEqualTo(5);
}

无限流场景下可调用StepVerifier的thenCancel()触发取消信号，取消信号也会触发对应metrics上报，不会出现指标丢失问题。如果是Spring Boot测试环境，也可以直接注入框架自动配置的MeterRegistry，注意在单测执行完成后清空注册表数据，避免不同用例之间的指标互相干扰。

内容的提问来源于stack exchange，提问作者Turan Yüksel