/*** JMX工具类*/publicclassJMXUtils {
privatefinalstaticStringJMX_CONNECT_URL="service:jmx:rmi:///jndi/rmi://%s/jmxrmi";
privatefinalstaticStringJMX_CONNECT_HOST="localhost:9999";
/*** 获取MBean attributes* @param mBeanName* @return*/publicstaticMap<String, Object>getMBeanAttrs(StringmBeanName) {
MBeanServerConnectionmbeanConnection=null;
Map<String, Object>result=newHashMap<>();
try (JMXConnectorconnector=JMXConnectorFactory                .connect(newJMXServiceURL(String.format(JMX_CONNECT_URL, JMX_CONNECT_HOST)))) {
mbeanConnection=connector.getMBeanServerConnection();
ObjectNamename=newObjectName(mBeanName);
MBeanInfomBeanInfo=mbeanConnection.getMBeanInfo(name);
for (MBeanAttributeInfoval : mBeanInfo.getAttributes()) {
Objectattribute=mbeanConnection.getAttribute(name, val.getName());
result.put(val.getName(), attribute);
            }
        } catch (Exceptione) {
e.printStackTrace();
        }
returnresult;
    }
publicstaticvoidmain(Stringargv[]) {
Map<String, Object>result=JMXUtils                .getMBeanAttrs("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec");
/*** 输出样例* name:RateUnit，valus:SECONDS* name:OneMinuteRate，valus:2.0212348203790318E-82* name:EventType，valus:bytes* name:FifteenMinuteRate，valus:1.2405687834521815E-6* name:Count，valus:351* name:FiveMinuteRate，valus:3.76580595627846E-17* name:MeanRate，valus:0.028921775433401117*/for (Map.Entry<String, Object>entry : result.entrySet()) {
System.out.println(String.format("name:%s，valus:%s", entry.getKey(), entry.getValue()));
        }
    }
}

publicclassPushReporterimplementsMetricsReporter {
privateHttpClienthttpClient;
@Overridepublicvoidconfigure(Map<String, ?>configs) {
// TODO Auto-generated method stub    }
@Overridepublicvoidinit(List<KafkaMetric>metrics) {
// 初始化HTTP客户端httpClient=HttpClient.newBuilder().build();
    }
@OverridepublicvoidmetricChange(KafkaMetricmetric) {
// 向自己的收集接口发送数据Stringurl="https://www.baidu.com";
try {
HttpRequestrequest=HttpRequest.newBuilder(newURI(url)).build();
HttpResponse<String>response=httpClient.send(request, HttpResponse.BodyHandlers.ofString());
System.out.println(String.format("metricName:%s, metricValue:%s, response:%s", metric.metricName().name(),
metric.metricValue(), response.statusCode()));
        } catch (IOException|InterruptedException|URISyntaxExceptione) {
e.printStackTrace();
        }
    }
@OverridepublicvoidmetricRemoval(KafkaMetricmetric) {
// TODO Auto-generated method stub    }
@Overridepublicvoidclose() {
// TODO Auto-generated method stub    }
}

publicclassKafkaConsumerClient {
publicstaticvoidmain(String[] args) {
Propertiesprops=newProperties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "test-consumer");
props.setProperty("enable.auto.commit", "false");
props.setProperty("auto.commit.interval.ms", "1000");
props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
// 上面实现的PushReporterprops.setProperty(ConsumerConfig.METRIC_REPORTER_CLASSES_CONFIG, "com.xdd.kafka.monitor.metric.PushReporter");
try (KafkaConsumer<String, String>consumer=newKafkaConsumer<>(props)) {
consumer.subscribe(Arrays.asList("test-topic"));
while (true) {
ConsumerRecords<String, String>records=consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord<String, String>record : records)
System.out.printf("offset = %d, key = %s, value = %s%n",
record.offset(),
record.key(),
record.value());
            }
        }
    }
}