Spring Boot Kafka Streams内置指标获取及自定义指标添加咨询

方案一：自定义实例级指标（代码实现） #

直接通过Kafka Streams的Metrics API采集线程级process-rate，聚合后暴露带统一标签的自定义指标，适配Spring Boot + Micrometer环境：

1. 注入核心依赖
   在Spring Boot配置类中注入MeterRegistry和StreamsBuilderFactoryBean，前者用于注册自定义指标，后者用于获取Kafka Streams实例。

2. 定时采集并聚合指标
   用@Scheduled定时（比如每秒）拉取所有线程的process-rate，计算实例的总/平均处理速率，再注册为带统一标签（如instance_id，用K8s Pod的HOSTNAME标识）的自定义指标：

@Configuration
@EnableScheduling
public class KafkaStreamsInstanceMetrics {

    @Autowired
    private MeterRegistry meterRegistry;

    @Autowired
    private StreamsBuilderFactoryBean streamsFactoryBean;

    @Scheduled(fixedRate = 1000)
    public void updateInstanceProcessRate() {
        KafkaStreams streams = streamsFactoryBean.getKafkaStreams();
        if (streams == null || !streams.state().isRunning()) {
            return;
        }

        // 筛选出线程级的process-rate指标
        List<Double> processRates = streams.metrics().metrics().stream()
                .filter(metric -> "stream-thread-metrics".equals(metric.metricName().group())
                        && "process-rate".equals(metric.metricName().name()))
                .map(metric -> (Double) metric.metricValue())
                .collect(Collectors.toList());

        // 计算总速率和平均速率
        double totalRate = processRates.stream().mapToDouble(Double::doubleValue).sum();
        double avgRate = processRates.isEmpty() ? 0 : totalRate / processRates.size();

        // 用K8s Pod的HOSTNAME作为实例唯一标识
        String instanceId = System.getenv("HOSTNAME");
        instanceId = instanceId == null ? UUID.randomUUID().toString() : instanceId;
        Tags tags = Tags.of("instance_id", instanceId);

        // 注册自定义指标到Micrometer
        Gauge.builder("kafka_streams_instance_total_process_rate", () -> totalRate)
                .tags(tags)
                .register(meterRegistry);
        Gauge.builder("kafka_streams_instance_avg_process_rate", () -> avgRate)
                .tags(tags)
                .register(meterRegistry);
    }
}

方案二：Prometheus端聚合（无需改代码） #

如果用Prometheus作为监控系统，直接通过配置对原始指标做标签清理和聚合，无需修改应用代码：

1. 标签清理
   在Prometheus的scrape_configs中添加relabel_configs，移除thread-id标签：

scrape_configs:
  - job_name: "kafka-streams-apps"
    kubernetes_sd_configs:
      - role: pod
        namespaces:
          names: [your-app-namespace]
    # 移除thread-id标签
    relabel_configs:
      - action: labeldrop
        regex: "thread-id"

2. 指标聚合查询
   查询时直接对实例维度聚合，比如求每个实例的总处理速率：

sum without (thread-id) (kafka_streams_stream_thread_metrics_process_rate)

或者求平均处理速率：

avg without (thread-id) (kafka_streams_stream_thread_metrics_process_rate)

方案三：自定义MetricsReporter（进阶） #

如果需要更灵活的指标控制，可以实现Kafka Streams的MetricsReporter接口，在指标生成时替换或统一标签。不过这种方式需要自定义实现并配置到Kafka Streams中，复杂度较高，适合有定制化需求的场景。

内容的提问来源于stack exchange，提问作者Danny.an