获取方法：
在 FlinkKafkaConsumer 获取 CPU 使用率，可以使用 getMetricGroup() 方法获取消费者的 metric group，然后使用 metric group 的 get(MetricName metricName) 方法获取 CPU 使用率指标

内部获取

import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;

public class MyKafkaConsumer extends FlinkKafkaConsumer<String> {
    public MyKafkaConsumer(String topic, DeserializationSchema<String> deserializer, Properties props) {
        super(topic, deserializer, props);
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        MetricGroup metricGroup = getMetricGroup();
        Counter cpuUsage = metricGroup.get(MetricNames.CPU_USAGE);
        double cpuUsageValue = cpuUsage.getCount() / (double) cpuUsage.getSum();

        System.out.println("CPU usage: " + cpuUsageValue);
    }
}

覆盖 FlinkKafkaConsumer 类的 open() 方法来获取 CPU 使用率指标。 getMetricGroup()方法用于获取消费者的metric group，metric group的get(MetricName metricName)方法用于获取CPU使用率metric。 Counter 的 getCount() 方法返回当前 CPU 使用率值，getSum() 方法返回最大可能值（通常为 1.0）。 要以百分比形式获得 CPU 使用率，我们将当前值除以最大值。 最后，将 CPU 使用率值打印到控制台。

外部获取

import org.apache.flink.metrics.Counter;
import org.apache.flink.metrics.MetricGroup;

FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("my-topic", new SimpleStringSchema(), properties);

MetricGroup metricGroup = consumer.getMetricGroup();
Counter cpuUsage = metricGroup.get(MetricNames.CPU_USAGE);
double cpuUsageValue = cpuUsage.getCount() / (double) cpuUsage.getSum();

System.out.println("CPU usage: " + cpuUsageValue);

获取表示 CPU 使用率指标的 Counter 对象。 Counter 的 getCount() 方法返回当前 CPU 使用率值，getSum() 方法返回最大可能值（通常为 1.0）。 要以百分比形式获得 CPU 使用率，我们将当前值除以最大值。