大数据-79 Kafka 集群模式 集群监控方案 JavaAPI获取集群指标 可视化监控集群方案： jconsole、Kafka Eagle

export KAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote \
                       -Dcom.sun.management.jmxremote.port=9999 \
                       -Dcom.sun.management.jmxremote.authenticate=false \
                       -Dcom.sun.management.jmxremote.ssl=false \
                       -Djava.rmi.server.hostname=${服务器的IP,尽量写IP,不要hostname或者域名}"

package icu.wzk.kafka;

import javax.management.MBeanServerConnection;
import javax.management.ObjectInstance;
import javax.management.ObjectName;
import javax.management.remote.JMXConnector;
import javax.management.remote.JMXConnectorFactory;
import javax.management.remote.JMXServiceURL;
import java.util.Iterator;
import java.util.Set;

public class JMXMonitorDemo {

    public static void main(String[] args) throws Exception {
        String jmxServiceUrl = "service:jmx:rmi:///jndi/rmi://h121.wzk.icu:9999/jmxrmi";
        JMXServiceURL jmxUrl = null;
        JMXConnector jmxc = null;
        MBeanServerConnection jmxs = null;
        ObjectName mbeanObjectName = null;
        Iterator sampleIter = null;
        Set sampleSet = null;

        // 创建JMXServiceURL 对象
        jmxUrl = new JMXServiceURL(jmxServiceUrl);
        // 建立指定的URL服务器的连接
        jmxc = JMXConnectorFactory.connect(jmxUrl);
        // 返回代表远程MBean服务器的MBeanServiceConnection对象
        jmxs = jmxc.getMBeanServerConnection();
        // 根据传入的字符串，创建ObjectName对象
        mbeanObjectName = new ObjectName("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec");
        // 指定ObjectName对应的MBeans
        sampleSet = jmxs.queryMBeans(null, mbeanObjectName);
        // 迭代器
        sampleIter = sampleSet.iterator();
        if (!sampleSet.isEmpty()) {
            // 如果返回了 则打印信息
            while (sampleIter.hasNext()) {
                ObjectInstance sampleObject = (ObjectInstance) sampleIter.next();
                ObjectName objectName = sampleObject.getObjectName();
                // 查看指定MBean指定属性的值
                String count = jmxs.getAttribute(objectName, "Count").toString();
                System.out.println("count: " + count);
            }
        }
        // 关闭
        jmxc.close();
    }
}

# Github 地址
# https://github.com/smartloli/EFAK

wget https://github.com/smartloli/kafka-eagle-bin/archive/v3.0.1.tar.gz
mv v3.0.1.tar.gz kafka-eagle-v3.0.1.tar.gz
tar -zxvf kafka-eagle-v3.0.1.tar.gz
cd kafka-eagle-bin-3.0.1/
tar -zxvf efak-web-3.0.1-bin.tar.gz
mv efak-web-3.0.1/ /opt/servers/

efak.zk.cluster.alias=cluster1
cluster1.zk.list=h121.wzk.icu:2181,h122.wzk.icu:2181,h123.wzk.icu:2181
######################################
# kafka sqlite jdbc driver address
######################################
efak.driver=org.sqlite.JDBC
efak.url=jdbc:sqlite:/hadoop/kafka-eagle/db/ke.db
efak.username=root
efak.password=www.kafka-eagle.org

# 我注释掉了MySQL

vim /etc/profile

# efak
export KE_HOME=/opt/servers/efak-web-3.0.1
export PATH=$PATH:$KE_HOME/bin

http://h121.wzk.icu:8048

admin
123456