kafka监控获取指定topic的消息总量示例
作者:扎克begod 发布时间:2023-09-04 01:44:48
标签:kafka,topic,消息总量
我就废话不多说了,直接 上代码吧!
import kafka.api.PartitionOffsetRequestInfo;
import kafka.common.TopicAndPartition;
import kafka.javaapi.OffsetResponse;
import kafka.javaapi.PartitionMetadata;
import kafka.javaapi.TopicMetadata;
import kafka.javaapi.TopicMetadataRequest;
import kafka.javaapi.consumer.SimpleConsumer;
import java.util.*;
import java.util.Map.Entry;
public class KafkaOffsetTools {
public final static String KAFKA_TOPIC_NAME_ADAPTER = "sample";
public final static String KAFKA_TOPIC_NAME_EXCEPTION = "exception";
public final static String KAFKA_TOPIC_NAME_AUDIT = "audit";
private static final String rawTopicTotal = "rawTopicTotalRecordCounter";
private static final String avroTopicTotal = "avroTopicTotalRecordCounter";
private static final String exceptionTopicTotal = "exceptionTopicTotalRecordCounter";
public KafkaOffsetTools() {
}
public static long getLastOffset(SimpleConsumer consumer, String topic,
int partition, long whichTime, String clientName) {
TopicAndPartition topicAndPartition = new TopicAndPartition(topic,
partition);
Map, PartitionOffsetRequestInfo> requestInfo = new HashMap, PartitionOffsetRequestInfo>();
requestInfo.put(topicAndPartition, new PartitionOffsetRequestInfo(
whichTime, 1));
kafka.javaapi.OffsetRequest request = new kafka.javaapi.OffsetRequest(
requestInfo, kafka.api.OffsetRequest.CurrentVersion(),
clientName);
OffsetResponse response = consumer.getOffsetsBefore(request);
if (response.hasError()) {
System.err.println("Error fetching data Offset Data the Broker. Reason: " + response.errorCode(topic, partition));
return 0;
}
long[] offsets = response.offsets(topic, partition);
return offsets[0];
}
private TreeMap, PartitionMetadata> findLeader(List a_seedBrokers, String a_topic) {
TreeMap, PartitionMetadata> map = new TreeMap, PartitionMetadata>();
loop:
for (String seed : a_seedBrokers) {
SimpleConsumer consumer = null;
try {
String[] hostAndPort;
hostAndPort = seed.split(":");
consumer = new SimpleConsumer(hostAndPort[0], Integer.valueOf(hostAndPort[1]), 100000, 64 * 1024,
"leaderLookup" + new Date().getTime());
List topics = Collections.singletonList(a_topic);
TopicMetadataRequest req = new TopicMetadataRequest(topics);
kafka.javaapi.TopicMetadataResponse resp = consumer.send(req);
List metaData = resp.topicsMetadata();
for (TopicMetadata item : metaData) {
for (PartitionMetadata part : item.partitionsMetadata()) {
map.put(part.partitionId(), part);
}
}
} catch (Exception e) {
System.out.println("Error communicating with Broker [" + seed
+ "] to find Leader for [" + a_topic + ", ] Reason: " + e);
} finally {
if (consumer != null)
consumer.close();
}
}
return map;
}
public static void main(String[] args) {
String kafkaBrokerList = System.getenv("metadata.broker.list");
if(kafkaBrokerList == null || kafkaBrokerList.length() == 0){
System.err.println("No config kafka metadata.broker.list,it is null .");
//for test
kafkaBrokerList = "localhost:9092,localhost:9093";
System.err.println("Use this broker list for test,metadata.broker.list="+kafkaBrokerList);
}
//init topic,logSize = 0
Map,Integer> topics = new HashMap,Integer>();
topics.put(KAFKA_TOPIC_NAME_ADAPTER,0);
topics.put(KAFKA_TOPIC_NAME_EXCEPTION,0);
topics.put(KAFKA_TOPIC_NAME_AUDIT,0);
//init kafka broker list
String[] kafkaHosts;
kafkaHosts = kafkaBrokerList.split(",");
if (kafkaHosts == null || kafkaHosts.length == 0) {
System.err.println("No config kafka metadata.broker.list,it is null .");
System.exit(1);
}
List seeds = new ArrayList();
for (int i = 0; i < kafkaHosts.length; i++) {
seeds.add(kafkaHosts[i]);
}
KafkaOffsetTools kot = new KafkaOffsetTools();
for(String topicName : topics.keySet()){
TreeMap, PartitionMetadata> metadatas = kot.findLeader(seeds, topicName);
int logSize = 0;
for (Entry, PartitionMetadata> entry : metadatas.entrySet()) {
int partition = entry.getKey();
String leadBroker = entry.getValue().leader().host();
String clientName = "Client_" + topicName + "_" + partition;
SimpleConsumer consumer = new SimpleConsumer(leadBroker, entry.getValue().leader().port(), 100000,
64 * 1024, clientName);
long readOffset = getLastOffset(consumer, topicName, partition,
kafka.api.OffsetRequest.LatestTime(), clientName);
logSize += readOffset;
if (consumer != null) consumer.close();
}
topics.put(topicName,logSize);
}
System.out.println(topics.toString());
System.out.println(rawTopicTotal+"="+topics.get(KAFKA_TOPIC_NAME_ADAPTER)+" "+System.currentTimeMillis());
System.out.println(avroTopicTotal+"="+topics.get(KAFKA_TOPIC_NAME_AUDIT)+" "+System.currentTimeMillis());
System.out.println(exceptionTopicTotal+"="+topics.get(KAFKA_TOPIC_NAME_EXCEPTION)+" "+System.currentTimeMillis());
}
}
来源:https://blog.csdn.net/aa5305123/article/details/88255435
0
投稿
猜你喜欢
- 回收站(Recycle Bin)从原理上来说就是一个数据字典表,放置用户删除(drop)掉的数据库对象信息。用户进行删除操作的对象并没有被数
- 本文实例讲述了彻底删除thinkphp3.1案例blog标签的方法。分享给大家供大家参考。具体方法如下:thinkphp3.1框架中的案例b
- 有如下 Pandas DataFrame:import pandas as pdinp = [{'c1':10, '
- 一、_func 单下划线开头 --口头私有变量1.1、在模块中使用单下划线开头在Python中,通过单下划线_来实现模块级别的私有化,变量除
- CSS处理斜角导航条的一个例子,这个是写着测试用的。暂没有实际的应用。斜角处理比较麻烦,主要有两个地方。1、图片处理。2、负数的理解。这两个
- 试了一下,xmlDoc.save()行不同,就试着用fso做了出来。整理一下,供大家discuss。由于用js操作本地xml文件之后save
- 开发web应用程序是一件非常辛苦的事情,你需要花大把大把的时间来做无数的事情。假如你不运用有条理的方法,尤其是在复杂的项目中,你会承受忽视项
- 阅读《YUI学习笔记(1)》YAHOO.lang.dump 与 YAHOO.lang.substitute。1.&nbs
- from urllib.request import urlopen  
- 代码案例import pluggy# HookspecMarker 和 HookimplMarker 实质上是一个装饰器带参数的装饰器类,作
- 在my.cnf中设置了sql_mode='STRICT_TRANS_TABLES';却发现还是有非法数据的写入,比如无符号整
- 有一台windows服务器上跑着mysql的一些应用,现在需要将mysql的数据每天备份,并通过ftp上传到指定的存储服务器上要是在linu
- 一、问题引发思考前阵子与同事探讨一个小需求时又遇到了按钮表示“动作”和表示“状态”间矛盾问题。想想这个问题多年前已经开始讨论了,所以在此整理
- 主要内容所谓RPC,是远程过程调用(Remote Procedure Call)的简写,网上解释很多,简单来说,就是在当前进程调用其他进程的
- 为什么我写ASP分页教程要提到AJAX呢,因为我们要多练习一下编程过程中,结构化的重要性. 再加上很多朋友对分页感到很高深,所以一直都不敢去
- 不知各位是否有手写代码的习惯。例如:要在一个单元格插入一段CSS代码,或者一段Javascript代码,怎么做才比较快捷方便呢?虽然Drea
- MySQL数据库线程缓冲池的相关知识是本文我们主要要介绍的内容,MySQL数据库支持线程缓存,在多线程连接模式下,如果连接断开后,将这个线程
- 一直在用JS写ASP,也不是特别原因,只是当初学的是JS,后来学ASP时知道ASP也可以用JS写,就没去学VBS.前几个月刚学ASP的时候找
- 一、IE透明度问题在IE的高度超过某一阀值时,会产生透明度不时失效的问题,这现象比较奇怪,(会有的时候全黑,有的时候全白)你有可能无法复现。
- 问题:pycharm无法调用pip安装的包原因:pycharm没有设置解析器解决方法:打开pycharm->File->Sett