后端代碼：

Monitor.java代碼：

package com.suncreate.kafkaConsumerMonitor.service;import com.suncreate.kafkaConsumerMonitor.model.ConsumerInfo;import org.apache.kafka.clients.consumer.ConsumerConfig;import org.apache.kafka.clients.consumer.KafkaConsumer;import org.apache.kafka.clients.consumer.OffsetAndMetadata;import org.apache.kafka.common.PartitionInfo;import org.apache.kafka.common.TopicPartition;import org.apache.kafka.common.serialization.StringDeserializer;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import java.text.DecimalFormat;import java.text.SimpleDateFormat;import java.util.*;/** * kafka消費(fèi)監(jiān)控 * * @author suxiang */public class Monitor { private static final Logger log = LoggerFactory.getLogger(Monitor.class); private String servers; private String topic; private String groupId; private long lastTime; private long lastTotalLag = 0L; private long lastLogSize = 0L; private long lastOffset = 0L; private double lastRatio = 0; private long speedLogSize = 0L; private long speedOffset = 0L; private String time; private List<ConsumerInfo> list; private final SimpleDateFormat simpleDateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss'); public String getTime() {return time; } public void setTime(String time) {this.time = time; } public long getLastTotalLag() {return lastTotalLag; } public double getLastRatio() {return lastRatio; } public String getTopic() {return topic; } public String getGroupId() {return groupId; } public long getSpeedLogSize() {return speedLogSize; } public long getSpeedOffset() {return speedOffset; } public List<ConsumerInfo> getList() {return list; } public void setList(List<ConsumerInfo> list) {this.list = list; } private KafkaConsumer<String, String> consumer; private List<TopicPartition> topicPartitionList; private final DecimalFormat decimalFormat = new DecimalFormat('0.00'); public Monitor(String servers, String topic, String groupId) {this.servers = servers;this.topic = topic;this.groupId = groupId;this.list = new ArrayList<>();//消費(fèi)者Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, this.servers);properties.put(ConsumerConfig.GROUP_ID_CONFIG, this.groupId);properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);consumer = new KafkaConsumer<String, String>(properties);//查詢 topic partitionstopicPartitionList = new ArrayList<>();List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);for (PartitionInfo partitionInfo : partitionInfoList) { TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition()); topicPartitionList.add(topicPartition);} } public void monitor(boolean addToList) {try { long startTime = System.currentTimeMillis(); //查詢 log size Map<Integer, Long> endOffsetMap = new HashMap<>(); Map<TopicPartition, Long> endOffsets = consumer.endOffsets(topicPartitionList); for (TopicPartition partitionInfo : endOffsets.keySet()) {endOffsetMap.put(partitionInfo.partition(), endOffsets.get(partitionInfo)); } //查詢消費(fèi) offset Map<Integer, Long> commitOffsetMap = new HashMap<>(); for (TopicPartition topicAndPartition : topicPartitionList) {OffsetAndMetadata committed = consumer.committed(topicAndPartition);commitOffsetMap.put(topicAndPartition.partition(), committed.offset()); } long endTime = System.currentTimeMillis(); log.info('查詢logSize和offset耗時(shí):' + (new DecimalFormat('0.000')).format((endTime - startTime) / 1000.0) + ' 秒'); startTime = System.currentTimeMillis(); //累加lag long totalLag = 0L; long logSize = 0L; long offset = 0L; if (endOffsetMap.size() == commitOffsetMap.size()) {for (Integer partition : endOffsetMap.keySet()) { long endOffset = endOffsetMap.get(partition); long commitOffset = commitOffsetMap.get(partition); long diffOffset = endOffset - commitOffset; totalLag += diffOffset; logSize += endOffset; offset += commitOffset;} } else {log.error('Topic:' + topic + ' consumer:' + consumer + ' topic partitions lost'); } log.info('Topic:' + topic + ' logSize:' + logSize + ' offset:' + offset + ' totalLag:' + totalLag); if (lastTime > 0) {if (System.currentTimeMillis() - lastTime > 0) { speedLogSize = (long) ((logSize - lastLogSize) / ((System.currentTimeMillis() - lastTime) / 1000.0)); speedOffset = (long) ((offset - lastOffset) / ((System.currentTimeMillis() - lastTime) / 1000.0));}if (speedLogSize > 0) { String strRatio = decimalFormat.format(speedOffset * 100 / (speedLogSize * 1.0)); lastRatio = Double.parseDouble(strRatio); log.info('Topic:' + topic + ' speedLogSize:' + speedLogSize + ' speedOffset:' + speedOffset + ' 百分比:' + strRatio + '%');} } lastTime = System.currentTimeMillis(); lastTotalLag = totalLag; lastLogSize = logSize; lastOffset = offset; endTime = System.currentTimeMillis(); log.info('計(jì)算耗時(shí):' + (new DecimalFormat('0.000')).format((endTime - startTime) / 1000.0) + ' 秒'); if (addToList) {this.setTime(simpleDateFormat.format(new Date()));this.list.add(new ConsumerInfo(this.getTopic(), this.getGroupId(), this.getLastTotalLag(), this.getLastRatio(), this.getSpeedLogSize(), this.getSpeedOffset(), this.getTime()));if (this.list.size() > 500) { this.list.remove(0);} }} catch (Exception e) { log.error('Monitor error', e);} }}

MonitorService.java代碼：

package com.suncreate.kafkaConsumerMonitor.service;import com.suncreate.kafkaConsumerMonitor.model.ConsumerInfo;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Value;import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;import java.util.*;@Servicepublic class MonitorService { private static final Logger log = LoggerFactory.getLogger(MonitorService.class); @Value('${kafka.consumer.servers}') private String servers; private Monitor monitor; private List<Monitor> monitorList; @PostConstruct private void Init() {monitorList = new ArrayList<>();monitorList.add(new Monitor(servers, 'wifiData', 'wifi-kafka-hbase'));monitorList.add(new Monitor(servers, 'KK_PASS_INFO_TYCC', 'EXTRACT-SAMPLE'));monitorList.add(new Monitor(servers, 'KK_PASS_INFO_TYCC', 'dblrecog-upload2vcn'));monitorList.add(new Monitor(servers, 'KK_PASS_INFO_TYCC_FILTER', 'yisa20210521000001'));monitorList.add(new Monitor(servers, 'KK_PASS_INFO_TYCC_FILTER', 'kafka-filter-check-19'));monitorList.add(new Monitor(servers, 'motorVehicle', 'unifiedstorage-downloader'));monitorList.add(new Monitor(servers, 'motorVehicle', 'full-vehicle-data-storage-kafka2ch'));monitorList.add(new Monitor(servers, 'motorVehicle', 'vehicle_store'));monitorList.add(new Monitor(servers, 'motorVehicle', 'vcn-sk-upload-luyang'));monitorList.add(new Monitor(servers, 'motorVehicle', 'vcn-sk-upload-yaohai'));monitorList.add(new Monitor(servers, 'motorVehicle', 'vcn-sk-upload-baohe'));monitorList.add(new Monitor(servers, 'peopleFace', 'kafka-filter-check-19')); } public void monitorOnce(boolean addToList) {for (Monitor monitor : monitorList) { monitor.monitor(addToList);} } public List<ConsumerInfo> getConsumerList() {List<ConsumerInfo> list = new ArrayList<>();for (Monitor monitor : monitorList) { list.add(new ConsumerInfo(monitor.getTopic(), monitor.getGroupId(), monitor.getLastTotalLag(), monitor.getLastRatio(), monitor.getSpeedLogSize(), monitor.getSpeedOffset(), monitor.getTime()));}return list; } public List<ConsumerInfo> getDetails(String topic, String groupId) {for (Monitor monitor : monitorList) { if (monitor.getTopic().equals(topic) && monitor.getGroupId().equals(groupId)) {return monitor.getList(); }}return new ArrayList<>(); }}

MonitorConfig.java代碼：

package com.suncreate.kafkaConsumerMonitor.task;import com.suncreate.kafkaConsumerMonitor.service.MonitorService;import org.slf4j.Logger;import org.slf4j.LoggerFactory;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Configuration;import org.springframework.scheduling.annotation.EnableScheduling;import org.springframework.scheduling.annotation.SchedulingConfigurer;import org.springframework.scheduling.config.ScheduledTaskRegistrar;import org.springframework.scheduling.support.CronTrigger;import java.text.SimpleDateFormat;@Configuration@EnableSchedulingpublic class MonitorConfig implements SchedulingConfigurer { private static final Logger logger = LoggerFactory.getLogger(MonitorConfig.class); private String cronExpression = '0 */3 * * * ?'; //private String cronExpression = '*/20 * * * * ?'; private final SimpleDateFormat simpleDateFormat = new SimpleDateFormat('yyyy-MM-dd HH:mm:ss.SSS'); @Autowired private MonitorService monitorService; @Override public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {taskRegistrar.addTriggerTask(() -> { monitorService.monitorOnce(true);}, triggerContext -> new CronTrigger(cronExpression).nextExecutionTime(triggerContext)); }}

MonitorController.java代碼：

package com.suncreate.kafkaConsumerMonitor.controller;import com.suncreate.kafkaConsumerMonitor.model.ConsumerInfo;import com.suncreate.kafkaConsumerMonitor.model.LayuiData;import com.suncreate.kafkaConsumerMonitor.service.MonitorService;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.web.bind.annotation.GetMapping;import org.springframework.web.bind.annotation.PostMapping;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;import java.util.List;@RestController@RequestMapping('/monitor')public class MonitorController { @Autowired private MonitorService monitorService; @GetMapping('/getConsumers') public LayuiData getConsumers() {List<ConsumerInfo> list = monitorService.getConsumerList();LayuiData data = new LayuiData(list);return data; } @GetMapping('/monitorOnce') public void monitorOnce() {monitorService.monitorOnce(false); } @GetMapping('/getDetails') public LayuiData getDetails(String topic, String groupId) {List<ConsumerInfo> list = monitorService.getDetails(topic, groupId);LayuiData data = new LayuiData(list);return data; }}

pom.xml文件(有些東西沒用到或者備用，沒有刪)：

<?xml version='1.0' encoding='UTF-8'?><project xmlns='http://maven.apache.org/POM/4.0.0' xmlns:xsi='http://www.w3.org/2001/XMLSchema-instance' xsi:schemaLocation='http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd'> <modelVersion>4.0.0</modelVersion> <parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.6.RELEASE</version><relativePath/> <!-- lookup parent from repository --> </parent> <groupId>com.suncreate</groupId> <artifactId>kafka-consumer-monitor</artifactId> <version>1.0</version> <name>kafka-consumer-monitor</name> <description>Kafka消費(fèi)積壓監(jiān)控預(yù)警</description> <properties><java.version>1.8</java.version><elasticsearch.version>6.1.4</elasticsearch.version> </properties> <dependencies><dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <version>1.18.12</version></dependency><dependency> <groupId>com.alibaba</groupId> <artifactId>fastjson</artifactId> <version>1.2.54</version></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope></dependency><dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka-test</artifactId> <scope>test</scope></dependency><dependency> <groupId>com.google.code.gson</groupId> <artifactId>gson</artifactId> <version>2.8.0</version></dependency><dependency> <groupId>org.postgresql</groupId> <artifactId>postgresql</artifactId> <scope>runtime</scope></dependency><dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-jdbc</artifactId></dependency><dependency> <groupId>org.elasticsearch.client</groupId> <artifactId>elasticsearch-rest-high-level-client</artifactId> <version>6.1.4</version></dependency><dependency> <groupId>com.oracle</groupId> <artifactId>ojdbc6</artifactId> <version>11.1.0.7.0</version></dependency><dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka_2.11</artifactId> <version>0.11.0.1</version></dependency><dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.11.0.1</version></dependency> </dependencies> <build><plugins> <plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId> </plugin> <plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration> <source>8</source> <target>8</target></configuration> </plugin></plugins> </build></project>

前端使用了 Layui 和 ECharts 展示表格和圖表

index.css代碼：

.div-title { font-size: 18px; margin-top: 10px; margin-left: 10px;}.div-right { text-align: right;}.span-red { color: #ff0000;}

index.html代碼(展示topic、消費(fèi)者組Consumer GroupId、Total Lag、Kafka數(shù)據(jù)生產(chǎn)速度、Kafka數(shù)據(jù)消費(fèi)速度等)：

<!DOCTYPE html><html lang='zh'><head> <meta charset='UTF-8'> <title>Title</title> <link rel='stylesheet' href='http://m.b3g6.com/bcjs/css/index.css' rel='external nofollow' rel='external nofollow' > <link rel='stylesheet' href='http://m.b3g6.com/bcjs/js/layui-v2.6.8/css/layui.css' rel='external nofollow' rel='external nofollow' media='all'> <script type='text/javascript' src='http://m.b3g6.com/bcjs/js/jquery-1.7.1.js'></script> <script type='text/javascript' src='http://m.b3g6.com/bcjs/js/layui-v2.6.8/layui.js' charset='utf-8'></script></head><body><div class='div-title'>Kafka 監(jiān)控 <button type='button' onclick='refreshTable()'>刷新</button></div><table id='myTable'></table><script type='text/javascript'> var myTable; layui.use(’table’, function () {var table = layui.table;myTable = table.render({ elem: ’#myTable’, url: ’/home/monitor/getConsumers’, cellMinWidth: 80, //全局定義常規(guī)單元格的最小寬度 cols: [[{field: ’topic’, width: 300, title: ’topic’, sort: true},{field: ’groupId’, width: 300, title: ’groupId’},{ field: ’totalLag’, width: 150, title: ’Total Lag’, sort: true, templet: function (d) {if (d.delayDay * 24 > 2) { return ’<div class='div-right'><span class='span-red'>’ + d.totalLag + ’</span></div>’} else { return ’<div class='div-right'><span>’ + d.totalLag + ’</span></div>’} }},{ field: ’speedLogSize’, width: 150, title: ’生產(chǎn)速度(條/秒)’, templet: function (d) {return ’<div class='div-right'>’ + d.speedLogSize + ’</div>’ }},{ field: ’speedOffset’, width: 150, title: ’消費(fèi)速度(條/秒)’, templet: function (d) {return ’<div class='div-right'>’ + d.speedOffset + ’</div>’ }},{ field: ’ratio’, width: 100, title: ’消費(fèi)/生產(chǎn)’, templet: function (d) {if (d.ratio < 90) { return ’<div class='div-right'><span class='span-red'>’ + d.ratio + ’%</span></div>’} else { return ’<div class='div-right'><span>’ + d.ratio + ’%</span></div>’} }},{ field: ’delayDay’, width: 150, title: ’積壓(天)’, sort: true, templet: function (d) {if (d.delayDay * 24 > 2) { return ’<div class='div-right'><span class='span-red'>’ + d.delayDay + ’</span></div>’} else { return ’<div class='div-right'><span>’ + d.delayDay + ’</span></div>’} }},{ field: ’ope’, width: 100, title: ’操作’, templet: function (d) {return ’<a href='http://m.b3g6.com/home/detail.html?topic=’ + d.topic + ’&groupId=’ + d.groupId + ’' rel='external nofollow' target='_blank' >詳細(xì)</a>’; }} ]]}); }); function refreshTable() {if (myTable) { myTable.reload();} } setInterval(function () {refreshTable(); }, 30000); // setInterval(function () { // $.get('/home/monitor/monitorOnce'); // }, 30000);</script></body></html>

detail.html代碼(展示單個(gè)消費(fèi)者組的Total Lag、生產(chǎn)速度、消費(fèi)速度以及Total Lag趨勢圖)：

<!DOCTYPE html><html lang='zh'><head> <meta charset='UTF-8'> <title>Title</title> <link rel='stylesheet' href='http://m.b3g6.com/bcjs/css/index.css' rel='external nofollow' rel='external nofollow' > <link rel='stylesheet' href='http://m.b3g6.com/bcjs/js/layui-v2.6.8/css/layui.css' rel='external nofollow' rel='external nofollow' media='all'> <script type='text/javascript' src='http://m.b3g6.com/bcjs/js/jquery-1.7.1.js'></script> <script type='text/javascript' src='http://m.b3g6.com/bcjs/js/layui-v2.6.8/layui.js' charset='utf-8'></script> <script type='text/javascript' src='http://m.b3g6.com/bcjs/js/echarts-v4.7.0/echarts.min.js'></script></head><body><div class='div-title'><span id='detailTitle'></span> 明細(xì) <button type='button' onclick='refreshTable()'>刷新</button></div><div style='height:400px;'></div><table id='test'></table><script type='text/javascript'> var myTable; var topic = getQueryVariable('topic'); var groupId = getQueryVariable('groupId'); $('#detailTitle').html(topic + ' ' + groupId); layui.use(’table’, function () {var table = layui.table;myTable = table.render({ elem: ’#test’, url: ’/home/monitor/getDetails?topic=’ + topic + ’&groupId=’ + groupId, cellMinWidth: 80, //全局定義常規(guī)單元格的最小寬度 initSort: {field: ’time’, //排序字段，對應(yīng) cols 設(shè)定的各字段名type: ’desc’ //排序方式 asc: 升序、desc: 降序、null: 默認(rèn)排序 }, cols: [[{field: ’topic’, width: 300, title: ’topic’},{field: ’groupId’, width: 300, title: ’groupId’},{field: ’time’, width: 180, title: ’時(shí)間’, sort: true},{ field: ’totalLag’, width: 150, title: ’Total Lag’, templet: function (d) {if (d.delayDay * 24 > 2) { return ’<div class='div-right'><span class='span-red'>’ + d.totalLag + ’</span></div>’} else { return ’<div class='div-right'><span>’ + d.totalLag + ’</span></div>’} }},{ field: ’speedLogSize’, width: 150, title: ’生產(chǎn)速度(條/秒)’, templet: function (d) {return ’<div class='div-right'>’ + d.speedLogSize + ’</div>’ }},{ field: ’speedOffset’, width: 150, title: ’消費(fèi)速度(條/秒)’, templet: function (d) {return ’<div class='div-right'>’ + d.speedOffset + ’</div>’ }},{ field: ’ratio’, width: 100, title: ’消費(fèi)/生產(chǎn)’, templet: function (d) {if (d.ratio < 90) { return ’<div class='div-right'><span class='span-red'>’ + d.ratio + ’%</span></div>’} else { return ’<div class='div-right'><span>’ + d.ratio + ’%</span></div>’} }},{ field: ’delayDay’, width: 150, title: ’積壓(天)’, templet: function (d) {if (d.delayDay * 24 > 2) { return ’<div class='div-right'><span class='span-red'>’ + d.delayDay + ’</span></div>’} else { return ’<div class='div-right'><span>’ + d.delayDay + ’</span></div>’} }} ]]}); }); function refreshTable() {if (myTable) { myTable.reload();}showChart(); } setInterval(function () {refreshTable(); }, 30000); function getQueryVariable(variable) {var query = window.location.search.substring(1);var vars = query.split('&');for (var i = 0; i < vars.length; i++) { var pair = vars[i].split('='); if (pair[0] == variable) {return pair[1]; }}return (false); } function showChart() {$.ajax({ type: 'GET', url: ’/home/monitor/getDetails?topic=’ + topic + ’&groupId=’ + groupId, success: function (data) {if (data && data.data && data.data.length > 1) { debugger; var chartDom = document.getElementById(’main’); var myChart = echarts.init(chartDom); var option; var xAxis = []; var serseis = []; for (var i = 0; i < data.data.length; i++) {xAxis.push(data.data[i].time);serseis.push(data.data[i].totalLag); } option = {title: { show: true, text: 'Total Lag 趨勢圖', x: ’center’},xAxis: { type: ’category’, data: xAxis},yAxis: { type: ’value’},series: [{ data: serseis, type: ’line’}] }; myChart.setOption(option);} }}); } showChart();</script></body></html>

效果圖：

消費(fèi)者組列表：

消費(fèi)者組明細(xì)：

到此這篇關(guān)于Java Kafka 消費(fèi)積壓監(jiān)控的文章就介紹到這了,更多相關(guān)Java Kafka 消費(fèi)監(jiān)控內(nèi)容請搜索好吧啦網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持好吧啦網(wǎng)！