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

以上这篇kafka监控获取指定topic的消息总量示例就是小编分享给大家的全部内容了,希望能给大家一个参考,也希望大家多多支持我们。

(0)

相关推荐

  • python每5分钟从kafka中提取数据的例子

    我就废话不多说了,直接上代码吧! import sys sys.path.append("..") from datetime import datetime from utils.kafka2file import KafkaDownloader import os """ 实现取kafka数据,文件按照取数据的间隔命名 如每5分钟从kafka取数据写入文件中,文件名为当前时间加5 """ TOPIC = "rtz

  • python 消费 kafka 数据教程

    1.安装python模块 pip install --user kafka-python==1.4.3 如果报错压缩相关的错尝试安装下面的依赖 yum install snappy-devel yum install lz4-devel pip install python-snappy pip install lz4 2.生产者 #!/usr/bin/env python # coding : utf-8 from kafka import KafkaProducer import json

  • Python测试Kafka集群(pykafka)实例

    生产者代码: # -* coding:utf8 *- from pykafka import KafkaClient host = 'IP:9092, IP:9092, IP:9092' client = KafkaClient(hosts = host) print client.topics # 生产者 topicdocu = client.topics['my-topic'] producer = topicdocu.get_producer() for i in range(100):

  • 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.TopicMetadataR

  • RocketMQ获取指定消息的实现方法(源码)

    概要 消息查询是什么? 消息查询就是根据用户提供的msgId从MQ中取出该消息 RocketMQ如果有多个节点如何查询? 问题:RocketMQ分布式结构中,数据分散在各个节点,即便是同一Topic的数据,也未必都在一个broker上.客户端怎么知道数据该去哪个节点上查? 猜想1:逐个访问broker节点查询数据 猜想2:有某种数据中心存在,该中心知道所有消息存储的位置,只要向该中心查询即可得到消息具体位置,进而取得消息内容 实际: 1.消息Id中含有消息所在的broker的地址信息(IP\Po

  • kafka rabbitMQ及rocketMQ队列的消息可靠性保证分析

    目录 1.消息丢失 1.生产者发送失败 2.消费者消费失败 3.队列因为自身体原因丢失数据 2.消息顺序 1.kafka 2.rocketMQ 3.rabbitMQ 3.消息重复 1.消息丢失 1.生产者发送失败 所有消息队列都可能发生的问题 生产者发送消息后,队列未成功接收(网络原因或其他)而生产者不知情,消息丢失 生产者发送消息后,队列接收成功->生产者确认,但消息并未持久化,队列崩溃,消息丢失 针对这类问题,三种消息队列都提供了生产者消息发送确认的模式,例如将kafka的acks参数设置为

  • PHP获取指定日期是星期几的实现方法

    本文实例讲述了PHP获取指定日期是星期几的实现方法.分享给大家供大家参考,具体如下: <?php header("Content-type: text/html; charset=utf-8"); //获取星期方法 function get_week($date){ //强制转换日期格式 $date_str=date('Y-m-d',strtotime($date)); //封装成数组 $arr=explode("-", $date_str); //参数赋值

  • Delphi实现获取句柄并发送消息的方法

    本文以实例形式详细说明了Delphi获取句柄并发送消息的方法,具体用法说明如下: 查找另外一个窗口的句柄: handle := FindWindow(nil,PChar('窗口的标题'));//查到窗体句柄 查找子窗体: childHandle := FindWindowEx(handle,0,'子窗体类','子窗体标题'); 另外有个枚举子窗体的API, EnumChildWindows(主创体句柄,@回调函数,用户参数); 用这个函数需要自己写一个回调的函数,比如: function Enu

  • js获取指定字符前/后的字符串简单实例

    如下所示: <!doctype html> <html> <head> <meta charset="utf-8"> <title>无标题文档</title> </head> <script type="text/javascript"> /* string 字符串; str 指定字符; split(),用于把一个字符串分割成字符串数组; split(str)[0],读取

  • python获取指定目录下所有文件名列表的方法

    本文实例讲述了python获取指定目录下所有文件名列表的方法.分享给大家供大家参考.具体实现方法如下: 这里python代码实现获取文件名列表的功能,可以指定文件中包含的字符,方便提取特定类型的文件名列表: # -*- coding: utf-8 -*- #~ #------------------------------------------------------------------ #~ module:wlab #~ Filename:wgetfilelist.py #~ Funct

  • java正则表达式获取指定HTML标签的指定属性值且替换的方法

    实例如下: public static String repDomain(String source, String domain, String element, String attr) { String img = ""; Pattern p_image; Matcher m_image; String regEx_img = "<" + element + "[^<>]*?\\s" + attr + "=['\

  • Java基于正则表达式获取指定HTML标签指定属性值的方法

    本文实例讲述了Java基于正则表达式获取指定HTML标签指定属性值的方法.分享给大家供大家参考,具体如下: 有时可能会有这样的需求,从HTML页面获取指定标签的指定属性值,可以通过第三方库解析来获取,但是这样相对比较麻烦! 如果使用正则表达式,那么就变得简单了.代码如下: package com.mmq.regex; import java.util.ArrayList; import java.util.List; import java.util.regex.Matcher; import

  • JavaScript获取指定元素位置的方法

    本文实例讲述了JavaScript获取指定元素位置的方法.分享给大家供大家参考.具体如下: 复制代码 代码如下: function showpane() {   var self = document.getElementById("eID");   var left = self.getBoundingClientRect().left + document.documentElement.scrollLeft;   var top = self.getBoundingClientR

随机推荐