非Spring Boot应用配置Micrometer将Spring Kafka JMX指标导入InfluxDB

我使用以下Spring Integration Kafka配置，通过JConsole可以看到生产者服务器的大量JMX指标：

<int-kafka:outbound-channel-adapter kafka-template="kafkaTemplate" id="kafkaOutboundChannelAdapter" 
    topic="learning-topic" channel="KafkaAdapterChan" 
    send-failure-channel="FailureChan">
</int-kafka:outbound-channel-adapter>

<bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate">
    <constructor-arg>
        <bean class="org.springframework.kafka.core.DefaultKafkaProducerFactory">
            <constructor-arg>
                <map>
                    <entry key="bootstrap.servers" value="xyz:9092" />
                    ... <!-- more producer properties -->
                </map>
            </constructor-arg>
        </bean>
    </constructor-arg>
</bean> 

在Spring Boot环境下，这些指标默认会自动接入InfluxDB的Micrometer注册表，但当前非Spring Boot环境下默认未启用。我已尝试创建以下Influx相关Bean：

<bean id="customInfluxConfigs" class="CustomImplemtation">
    <!-- all the properties -->
</bean>

<bean id="influxRegistry" class="io.micrometer.influx.InfluxRegistry">
    <constructor-arg name="config" ref="customInfluxConfigs"/>
    <constructor-arg name="clock" ref="micrometerClock"/>
</bean>

但这样并未成功上报现有的默认JMX指标，同时我也尝试了以下代码：

private DefaultKafkaProducerFactory defaultProducerFactory;
private InfluxMeterRegistry influxMeterRegistry;

@Override
public void start(){
    defaultProducerFactory.createProducer();
    defaultProducerFactory.addListener(new MicrometerProducerListener(influxMeterRegistry));
}

现在需要实现通过Micrometer将JMX指标上报到InfluxDB，以便在Grafana中使用，寻求帮助。

1. 补全核心依赖 #

确保项目中包含以下依赖（根据构建工具调整，以Maven为例）：

<!-- Spring Kafka -->
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
    <version>你的对应版本</version>
</dependency>
<!-- Micrometer核心 -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-core</artifactId>
    <version>1.12.x</version>
</dependency>
<!-- InfluxDB适配 -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-influx</artifactId>
    <version>1.12.x</version>
</dependency>
<!-- JMX指标抓取 -->
<dependency>
    <groupId>io.micrometer</groupId>
    <artifactId>micrometer-jmx</artifactId>
    <version>1.12.x</version>
</dependency>

2. 手动配置JMX指标转发到InfluxDB #

非Spring Boot环境下，需要手动注册JMX指标抓取器，并将指标同步到InfluxRegistry：

import io.micrometer.core.instrument.Clock;
import io.micrometer.core.instrument.MeterRegistry;
import io.micrometer.core.instrument.binder.jmx.JmxMeterRegistry;
import io.micrometer.influx.InfluxConfig;
import io.micrometer.influx.InfluxMeterRegistry;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class MonitoringConfig {

    @Bean
    public InfluxConfig influxConfig() {
        return new InfluxConfig() {
            @Override
            public String get(String key) {
                // 按需返回Influx配置项，例如：
                if (key.equals("influx.url")) return "http://你的Influx地址:8086";
                if (key.equals("influx.username")) return "admin";
                if (key.equals("influx.password")) return "password";
                return null;
            }

            @Override
            public String db() {
                return "kafka_monitor"; // 指定存储指标的数据库名
            }
        };
    }

    @Bean
    public MeterRegistry influxMeterRegistry(InfluxConfig config, Clock clock) {
        return new InfluxMeterRegistry(config, clock);
    }

    @Bean
    public JmxMeterRegistry jmxMeterRegistry(MeterRegistry influxRegistry, Clock clock) {
        JmxMeterRegistry jmxRegistry = new JmxMeterRegistry(JmxConfig.DEFAULT, clock);
        // 将JMX抓取的指标转发到InfluxDB注册表
        influxRegistry.config().meterFilter(jmxRegistry.getMeterFilter());
        jmxRegistry.start();
        return jmxRegistry;
    }
}

3. 修复Kafka生产者指标绑定 #

你之前的代码顺序有误，需要先添加监听器，再创建生产者，否则已生成的生产者实例不会绑定指标监听：

@Override
public void start(){
    // 先绑定监听器
    defaultProducerFactory.addListener(new MicrometerProducerListener<>(influxMeterRegistry));
    // 再创建生产者
    defaultProducerFactory.createProducer();
}

4. 验证指标上报 #

• 启动应用后，登录InfluxDB查看是否生成kafka.producer、jmx.*开头的measurement
• 若指标未上报，开启Micrometer debug日志排查：

  <logger name="io.micrometer" level="DEBUG"/>
  

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