kafka监控获取指定topic的消息总量示例
作者:扎克begod 时间:2023-09-04 01:44:48
我就废话不多说了,直接 上代码吧!
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
标签:kafka,topic,消息总量
![](/images/zang.png)
![](/images/jiucuo.png)
猜你喜欢
如何用CocosCreator制作微信小游戏
2023-08-23 16:00:02
![](https://img.aspxhome.com/file/2023/7/56117_0s.png)
Vue生产和开发环境如何切换及过滤器的使用
2024-05-09 15:20:31
Pytorch环境搭建与基本语法
2021-04-22 21:57:47
![](https://img.aspxhome.com/file/2023/4/105894_0s.png)
Python命令行参数解析模块optparse使用实例
2023-11-04 08:09:08
Python数据结构与算法之完全树与最小堆实例
2021-01-10 06:58:47
Python离线安装openpyxl模块的步骤
2021-08-10 16:04:04
![](https://img.aspxhome.com/file/2023/6/89286_0s.png)
Golang如何交叉编译各个平台的二进制文件详解
2024-05-22 17:48:47
![](https://img.aspxhome.com/file/2023/6/125086_0s.png)
asp之自动闭合HTML/ubb标签函数+简单注释
2008-09-29 12:21:00
封装了一个Java数据库访问管理类
2024-01-25 04:52:15
浅谈Pytorch torch.optim优化器个性化的使用
2023-12-19 08:47:12
python识别验证码图片实例详解
2022-02-13 10:23:53
![](https://img.aspxhome.com/file/2023/4/95124_0s.png)
教你用python实现自动回复QQ消息(不到60行)
2021-05-17 06:19:20
![](https://img.aspxhome.com/file/2023/9/120879_0s.png)
关于换行和回车的图文小结
2023-07-17 14:41:37
![](https://img.aspxhome.com/file/2023/6/132326_0s.jpg)
基于opencv对高空拍摄视频消抖处理方法
2021-09-29 22:25:41
![](https://img.aspxhome.com/file/2023/9/133789_0s.gif)
CSS框架的相关汇总(CSS Frameworks)
2008-04-02 12:00:00
SQL Server"错误 21002: [SQL-DMO]用户 * 已经存在问题解决
2024-01-16 16:01:55
MySQL中ROUND函数进行四舍五入操作陷阱分析
2024-01-15 04:02:04
Bootstrap modal使用及点击外部不消失的解决方法
2024-04-10 13:56:45
MySql之授权用户权限如何设置
2024-01-21 21:56:13
![](https://img.aspxhome.com/file/2023/4/67594_0s.png)
多次执行mysql_fetch_array()的指针归位问题探讨
2024-01-16 00:16:16