首页 / JAVA / Java Kafka 消费积压监控

Java Kafka 消费积压监控

内容导读

互联网集市收集整理的这篇技术教程文章主要介绍了Java Kafka 消费积压监控，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含20935字，纯文字阅读大概需要30分钟。

内容图文

Java Kafka 消费积压监控

后端代码：

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消费监控
 *
 * @author suxiang
 */publicclass Monitor {
    privatestaticfinal Logger log = LoggerFactory.getLogger(Monitor.class);

    private String servers;

    private String topic;

    private String groupId;

    privatelong lastTime;

    privatelong lastTotalLag = 0L;

    privatelong lastLogSize = 0L;

    privatelong lastOffset = 0L;

    privatedouble lastRatio = 0;

    privatelong speedLogSize = 0L;

    privatelong speedOffset = 0L;

    private String time;

    private List<ConsumerInfo> list;

    privatefinal SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

    public String getTime() {
        return time;
    }

    publicvoid setTime(String time) {
        this.time = time;
    }

    publiclong getLastTotalLag() {
        return lastTotalLag;
    }

    publicdouble getLastRatio() {
        return lastRatio;
    }

    public String getTopic() {
        return topic;
    }

    public String getGroupId() {
        return groupId;
    }

    publiclong getSpeedLogSize() {
        return speedLogSize;
    }

    publiclong getSpeedOffset() {
        return speedOffset;
    }

    public List<ConsumerInfo> getList() {
        return list;
    }

    publicvoid setList(List<ConsumerInfo> list) {
        this.list = list;
    }

    private KafkaConsumer<String, String> consumer;

    private List<TopicPartition> topicPartitionList;

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

        //消费者
        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 partitions
        topicPartitionList = new ArrayList<>();
        List<PartitionInfo> partitionInfoList = consumer.partitionsFor(topic);
        for (PartitionInfo partitionInfo : partitionInfoList) {
            TopicPartition topicPartition = new TopicPartition(partitionInfo.topic(), partitionInfo.partition());
            topicPartitionList.add(topicPartition);
        }
    }

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

            //查询消费 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耗时:" + (new DecimalFormat("0.000")).format((endTime - startTime) / 1000.0) + " 秒");
            startTime = System.currentTimeMillis();

            //累加laglong 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("计算耗时:" + (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);
        }
    }

}

View Code

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.*;

@Service
publicclass MonitorService {
    privatestaticfinal Logger log = LoggerFactory.getLogger(MonitorService.class);

    @Value("${kafka.consumer.servers}")
    private String servers;

    private Monitor monitor;

    private List<Monitor> monitorList;

    @PostConstruct
    privatevoid 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"));
    }

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

}

View Code

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
@EnableScheduling

                public
                class MonitorConfig implements SchedulingConfigurer {
    privatestaticfinal Logger logger = LoggerFactory.getLogger(MonitorConfig.class);

    private String cronExpression = "0 */3 * * * ?";

    //private String cronExpression = "*/20 * * * * ?";privatefinal SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");

    @Autowired
    private MonitorService monitorService;

    @Override
    publicvoid configureTasks(ScheduledTaskRegistrar taskRegistrar) {
        taskRegistrar.addTriggerTask(() -> {

            monitorService.monitorOnce(true);

        }, triggerContext -> new CronTrigger(cronExpression).nextExecutionTime(triggerContext));
    }
}

View Code

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")
publicclass MonitorController {

    @Autowired
    private MonitorService monitorService;

    @GetMapping("/getConsumers")
    public LayuiData getConsumers() {

        List<ConsumerInfo> list = monitorService.getConsumerList();

        LayuiData data = new LayuiData(list);
        return data;
    }

    @GetMapping("/monitorOnce")
    publicvoid 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;
    }
}

View Code

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消费积压监控预警</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>

View Code

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

index.css代码：

                .div-title {
    font-size: 18px;
    margin-top: 10px;
    margin-left: 10px;
}

.div-right {
    text-align: right;
}

.span-red {
    color: #ff0000;
}

View Code

index.html代码(展示topic、消费者组Consumer GroupId、Total Lag、Kafka数据生产速度、Kafka数据消费速度等)：

                <!
                DOCTYPE html
                >
                <
                html 
                lang
                ="zh"
                >
                <
                head
                >
                <
                meta 
                charset
                ="UTF-8"
                >
                <
                title
                >Title</title><link rel="stylesheet" href="css/index.css"><link rel="stylesheet" href="js/layui-v2.6.8/css/layui.css" media="all"><script type="text/javascript" src="js/jquery-1.7.1.js"></script><script type="text/javascript" src="js/layui-v2.6.8/layui.js" charset="utf-8"></script></head><body><div class="div-title">Kafka 监控
    <button type="button" class="layui-btn layui-btn-sm" onclick="refreshTable()">刷新</button></div><table class="layui-hide" 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, //全局定义常规单元格的最小宽度            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: ‘生产速度(条/秒)‘, templet: function (d) {
                        return‘<div class="div-right">‘+ d.speedLogSize +‘</div>‘
                    }
                },
                {
                    field: ‘speedOffset‘, width: 150, title: ‘消费速度(条/秒)‘, templet: function (d) {
                        return‘<div class="div-right">‘+ d.speedOffset +‘</div>‘
                    }
                },
                {
                    field: ‘ratio‘, width: 100, title: ‘消费/生产‘, 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="/home/detail.html?topic=‘+ d.topic +‘&groupId=‘+ d.groupId +‘" target="_blank" class="layui-btn layui-btn-sm" >详细</a>‘;
                    }
                }
            ]]
        });
    });

    function refreshTable() {
        if (myTable) {
            myTable.reload();
        }
    }

    setInterval(function () {
        refreshTable();
    }, 30000);

    // setInterval(function () {//     $.get("/home/monitor/monitorOnce");// }, 30000);</script></body></html>

View Code

detail.html代码(展示单个消费者组的Total Lag、生产速度、消费速度以及Total Lag趋势图)：

                <!
                DOCTYPE html
                >
                <
                html 
                lang
                ="zh"
                >
                <
                head
                >
                <
                meta 
                charset
                ="UTF-8"
                >
                <
                title
                >Title</title><link rel="stylesheet" href="css/index.css"><link rel="stylesheet" href="js/layui-v2.6.8/css/layui.css" media="all"><script type="text/javascript" src="js/jquery-1.7.1.js"></script><script type="text/javascript" src="js/layui-v2.6.8/layui.js" charset="utf-8"></script><script type="text/javascript" src="js/echarts-v4.7.0/echarts.min.js"></script></head><body><div class="div-title"><span id="detailTitle"></span> 明细
    <button type="button" class="layui-btn layui-btn-sm" onclick="refreshTable()">刷新</button></div><div id="main" style="height:400px;"></div><table class="layui-hide" 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, //全局定义常规单元格的最小宽度            initSort: {
                field: ‘time‘, //排序字段，对应 cols 设定的各字段名                type: ‘desc‘//排序方式  asc: 升序、desc: 降序、null: 默认排序            },
            cols: [[
                {field: ‘topic‘, width: 300, title: ‘topic‘},
                {field: ‘groupId‘, width: 300, title: ‘groupId‘},
                {field: ‘time‘, width: 180, title: ‘时间‘, 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: ‘生产速度(条/秒)‘, templet: function (d) {
                        return‘<div class="div-right">‘+ d.speedLogSize +‘</div>‘
                    }
                },
                {
                    field: ‘speedOffset‘, width: 150, title: ‘消费速度(条/秒)‘, templet: function (d) {
                        return‘<div class="div-right">‘+ d.speedOffset +‘</div>‘
                    }
                },
                {
                    field: ‘ratio‘, width: 100, title: ‘消费/生产‘, 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>

View Code

效果图：

消费者组列表：

技术分享图片

消费者组明细：

技术分享图片

原文：https://www.cnblogs.com/s0611163/p/14957216.html

内容总结

以上是互联网集市为您收集整理的Java Kafka 消费积压监控全部内容，希望文章能够帮你解决Java Kafka 消费积压监控所遇到的程序开发问题。如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。

内容备注

版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容，请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。

内容手机端

扫描二维码推送至手机访问。

本文链接：https://qyyshop.com/info/1261408.html

来源：【匿名】

首页 / JAVA / Java Kafka 消费积压监控

Java Kafka 消费积压监控

内容导读

内容图文

内容总结

内容备注

内容手机端

【Java Kafka 消费积压监控】教程文章相关的互联网学习教程文章

Java Kafka 消费积压监控【代码】【图】

使用java创建kafka的生产者和消费者【代码】

java实现kafka发送消息和接收消息（java无注解方式+springBoot注解方式）【代码】【图】

kafka2.5.0生产者与消费者，java普通main方法示例【代码】

java-kafka安装以及使用案例【代码】【图】

Java 连接Kafka报错java.nio.channels.ClosedChannelExcep【代码】

Kafka使用Java客户端进行访问【代码】

如何获取 Kafka 的消费者详情 —— 从 Scala 到 Java 的切换【代码】【图】

kafka学习总结007 --- 生产者Java API实例【代码】【图】

2021年Java高频面试题分享（中高级版）：JVM+Redis+Kafka +数据库+设计模式【图】

BATJ高级Java面试题分享：JVM+Redis+Kafka +数据库+设计模式【图】

大数据架构开发挖掘分析 Hadoop HBase Hive Storm Spark Sqoop Flume ZooKeeper Kafka Redis MongoDB 机器学习云计算视频教程 Java互联网架构师【图】

京东三面凉凉：java+spring+jvm+kafka+微服务等一个都讲不清【图】

阿里架构师纯手写java面试宝典：spring/Redis/数据/Kafka/微服务，就问还有谁？【图】

java版spring cloud微服务架构b2b2c电子商务平台-kafka处理超大消息【代码】

KAFKA - 相关标签

JAVA - 技术教程分类

JAVA - 最新教程

JAVA - 最热教程