@InterceptInstantiationViaReflection(
        reporterClassName = "org.apache.flink.metrics.kafka.KafkaReporter")
public class KafkaReporterFactory implements MetricReporterFactory {

    @Override
    public MetricReporter createMetricReporter(Properties properties) {
        return new KafkaReporter();
    }
}

@Override
public void open(MetricConfig metricConfig) {
    LOG.info("metricConfig：" + metricConfig);
    topic = metricConfig.getString("topic", "");
    if (StringUtils.isBlank(topic)) {
        LOG.error("metrics.reporter.kafka_reporter.topic is null");
    }
    String endsWithMetric = metricConfig.getString("endswith.metricname", "").trim(); //指定需要獲取指標(biāo)名稱
    endsWithMetricList = Arrays.asList(endsWithMetric.split(","));
    String bootstrapservers = metricConfig.getString("bootstrap.servers", "");
    if (StringUtils.isBlank(bootstrapservers)) {
        LOG.error("metrics.reporter.kafka_reporter.bootstrap.servers is null");
    }
    Properties properties = new Properties();
    properties.put("bootstrap.servers", bootstrapservers);
    properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.put(
            "value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    kafkaProducer = new KafkaProducer<String, String>(properties);
}

@Override
public void report() {
    tryReport();
}
private final ObjectMapper mapper = new ObjectMapper();
private void tryReport() {
    Instant timestamp = Instant.now();
    try {
        String job_id = "";
        String job_name = "";
        List metriclist = new ArrayList<>();
        metriclist.addAll(gauges.values()); //獲取gauges類型指標(biāo)集
        metriclist.addAll(counters.values());//獲取gauges類型指標(biāo)集
        metriclist.addAll(histograms.values());//獲取histograms類型指標(biāo)集
        metriclist.addAll(meters.values());//獲取meters類型指標(biāo)集
//每個(gè)指標(biāo)數(shù)據(jù)里面都加上對(duì)應(yīng)的job_id, job_name
        for (MeasurementInfo info : metriclist) {
            if (info.getName().startsWith("jobmanager_job_")
                    || info.getName().startsWith("taskmanager_job_")) {
                job_id = info.getTags().getOrDefault(JOB_ID, "");
                job_name = info.getTags().getOrDefault(JOB_NAME, "");
                if (StringUtils.isBlank(job_id) || StringUtils.isBlank(job_name)) {
                    LOG.error("do not get job_id or job name:{}", info);
                }
                break;
            }
        }
        List

[
  {
    "name": "jobmanager.uptime",
    "time": 1647314569119,
    "fields": {
      "value": 1703478823
    },
    "tags": {
      "host": "bigdata-03",
      "job_id": "dc7a58b3f202059cd72c3467ecedb4b7",
      "job_name": "amp_zabbix_pre"
    }
  },
  {
    "name": "jobmanager.Status.JVM.Memory.Heap.Max",
    "time": 1647314569119,
    "fields": {
      "value": 468713472
    },
    "tags": {
      "host": "bigdata-03",
      "job_id": "",
      "job_name": ""
    }
  }，
  ...
]