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

kafka

34 浏览量更新于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页未读，立即下载

weixin_38504417

粉丝: 5
资源: 937

会员权益专享

kafka监控获取指定topic的消息总量示例

评论0

会员权益专享

最新资源

kafka监控获取指定topic的消息总量示例

评论0

kafka的topic小工具

kafka消息监控(linux运行_window查看)

kafka demo ，两种线程消费方式

springboot如何接受指定topic的kafka消息

java不订阅kafka topic获取topic偏移量

debezuim kafka 指定 topic

Springboot集成kafka消费指定的topic后讲消息发送到另一个topic上

Springboot集成kafka消费指定的topic

kafka java动态获取topic并动态创建消费者

kafka查看指定topic消费情况

linux 中kafka消费者指定topic分区

kafka2.4.1 获取 Topic数据大小，java代码

Kafka和MySQL的组合使用代码示例

springboot kafka 获取topoic

kafka golang 删除指定key消息

flink kafkasource获取 metadata

kafka定时获取信息

kafka创建log的topic

kafka 生产者消费者 收发消息代码示例

会员权益专享

最新资源

kafka 生产者消费者收发消息代码示例