kafka监控获取指定topic的消息总量示例_kafkaapi查看topic的消费组 - CSDN文库

178 浏览量更新于2023-03-03 评论收藏 35KB PDF 举报

身份认证购VIP最低享 7 折!

领优惠券(最高得80元）

资源详情

资源评论

资源推荐

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);

本内容试读结束，登录后可阅读更多

下载后可阅读完整内容，剩余1页未读，立即下载

评论0

weixin_38504417

粉丝: 5
资源: 937

会员权益专享

图片转文字

全年可省5，000元立即开通

最新资源

资源上传下载、课程学习等过程中有任何疑问或建议，欢迎提出宝贵意见哦~我们会及时处理！点击此处反馈