开启JMX

kafka开启JMX的2种方式：
1. 启动kafka时增加JMX_PORT=9988，即JMX_PORT=9988 bin/kafka-server-start.sh -daemon config/server.properties
2. 修改kafka-run-class.sh脚本，第一行增加JMX_PORT=9988即可。

事实上这两种配置方式背后的原理是一样的，我们看一下kafka的启动脚本kafka-server-start.sh的最后一行exec $base_dir/kafka-run-class.sh $EXTRA_ARGS kafka.Kafka "$@"，实际上就是调用kafka-run-class.sh脚本，其中有一段这样的内容：

# JMX port to use
if [  $JMX_PORT ]; thenKAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
fi

所以，本质是给参数JMX_PORT赋值，第二种方式在脚本的第一行增加JMX_PORT=9988，$JMX_PORT就能取到值；而第一种方式有点逼格，本质是设置环境变量然后执行启动脚本，类似下面这种方式给JMX_PORT赋值：

[afei@kafka]$ export JMX_PORT=9988
[afei@kafka]$ bin/kafka-server-start.sh -daemon config/server.properties

jmx所有相关参数都在脚本kafka-run-class.sh中，如下所示：

# JMX settings
if [ -z "$KAFKA_JMX_OPTS" ]; thenKAFKA_JMX_OPTS="-Dcom.sun.management.jmxremote -Djava.rmi.server.hostname=10.0.55.229 -Dcom.sun.management.jmxremote.authenticate=false  -Dcom.sun.management.jmxremote.ssl=false "
fi# JMX port to use
if [  $JMX_PORT ]; thenKAFKA_JMX_OPTS="$KAFKA_JMX_OPTS -Dcom.sun.management.jmxremote.port=$JMX_PORT "
fi

某些服务器可能无法正确绑定ip，这时候我们需要显示指定绑定的host：-Djava.rmi.server.hostname=10.0.55.229；

jconsole连接

配置好jmx并启动kafka后，可以启动jconsole验证jmx配置是否正确（连接远程进程的host就是参数java.rmi.server.hostname指定的值，port就是参数JMX_PORT指定的值）：

打开jconsole的控制台后，点击“MBean”这个Tab页，可以看到很多”kafka.”开头的属性：

程序获取

上面介绍了通过jconsole获取kafka注册到MBean中属性值。实际使用中，很可能有一些自定义监控kafka的研发，这时候就需要通过代码获取注册到MBean中属性值，MBean中的属性可以通过jconsole中得到，也可以通过kafka官方文档得到，请戳Kafka Monitor，获取jmx属性值的核心业务代码如下：

/*** @author afei* @version 1.0.0* @since 2018年06月19日*/
public class KafkaJmxConnection {private Logger log = LoggerFactory.getLogger(this.getClass());private MBeanServerConnection conn;/*** 默认连接的ip和端口号*/private String ipAndPort = "localhost:9999";public KafkaJmxConnection(String ipAndPort){this.ipAndPort = ipAndPort;}public boolean init() throws Exception {String jmxURL = "service:jmx:rmi:///jndi/rmi://" +ipAndPort+ "/jmxrmi";log.info("init jmx, jmxUrl: {}, and begin to connect it",jmxURL);try {// 初始化连接jmxJMXServiceURL serviceURL = new JMXServiceURL(jmxURL);JMXConnector connector = JMXConnectorFactory.connect(serviceURL, null);conn = connector.getMBeanServerConnection();if(conn == null){log.error("getValue connection return null!");return  false;}} catch (Exception e) {e.printStackTrace();return false;}return true;}public String getName(String metric, String topicName){if (topicName==null){return metric;}// 一些metric与topic有关, 如果要指定具体的topic的话return metric + ",topic=" + topicName;}public Map<String, Object> getValue(String topicName, String metric, Collection<String> attrs){ObjectName objectName;try {objectName = new ObjectName(this.getName(metric, topicName));} catch (MalformedObjectNameException e) {e.printStackTrace();return null;}Map<String, Object> result = new HashMap<>();// 遍历所有属性, 获取每个属性的结果, 并将属性和属性对应的结果保存到map中for(String attr:attrs){result.put(attr, getAttribute(objectName, attr));}return result;}public Map<String, Object> getValue(String metric, Collection<String> attrs){return getValue(null, metric, attrs);}private Object getAttribute(ObjectName objName, String objAttr){if(conn== null){log.error("jmx connection is null");return null;}try {return conn.getAttribute(objName,objAttr);} catch (Exception e) {e.printStackTrace();return null;}}}

从kafka官网摘取两个bean name：
- kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec
kafka官方文档的解释：Byte in rate from clients。
属性列表：从jconsole上可以看到BytesInPerSec包含的属性有Count、MeanRate、RateUnit、EventType、OneMinuteRate、FiveMinuteRate、FifteenMinuteRate。

kafka.server:type=ReplicaManager,name=PartitionCount
kafka官方文档的解释：Partition counts。
属性列表：Value

下面的代码就以获取kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec和kafka.server:type=ReplicaManager,name=PartitionCount为例：

import com.afei.kafka.metric.util.KafkaJmxConnection;
import com.google.common.collect.Lists;/*** @author afei* @version 1.0.0* @since 2018年06月19日*/
public class KafkaMetricMain {public static void main(String[] args) throws Exception {KafkaJmxConnection jmxConn = new KafkaJmxConnection("10.0.55.229:9988");jmxConn.init();while(true) {String topicName = "TPC_WALLET_UNFREEZE_DEDUCT_COMPENSATE";// 与topic无关的metricObject o1 = jmxConn.getValue("kafka.server:type=ReplicaManager,name=PartitionCount",Lists.newArrayList("Value"));System.out.println(o1);// 与topic有关的metricObject o2 = jmxConn.getValue(topicName,"kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec",Lists.newArrayList("Count", "OneMinuteRate", "FiveMinuteRate"));System.out.println(o2);Thread.sleep(5000);}}
}

3. kafka开启JMX相关推荐

【Kafka】Window下kafka开启JMX监控
1.概述因为需要,需要在windows下开启kafka,然后kafka开启JMX监控同样是修改kafka-server-start文件,但是修改的是kafka-server-start.bat I ...
【kafka】kafka docker jmx 远程连接跨容器
1.概述前情提要: [Java]java jmx 入门案例 [kafka]java使用jmx 监控Kafka [Kafka]Window下kafka开启JMX监控 2.远程连接 kafka jxm远 ...
Kafka 开启 SASL/PLAINTEXT 认证及 ACL
Linux 安装 Kafka 并开启 SASL/PLAINTEXT 认证前言一.环境准备 1.组件版本 2.下载文件 3.上传文件二.安装 Zookeeper(单节点) 三.安装 Kafka(单 ...
java jmx 监控tomcat_通过Tomcat开启JMX监控的方法图解
搭建模拟环境: 操作系统:centos7 内存:1G jdk:1.8.0_131 tomcat:8.0.48 环境准备我们这里就不直接演示了,直接配置tomcat的jmx 1.进入到tomcat的bi ...
为Java应用开启JMX服务
. 前言 JMX(Java Management Extensions)这项技术可用于监管Java应用.通常是由被监管程序开启JMX服务,再通过 JConsole.JMC.VisualVM之类的工具利 ...
java jmx 开启_动态开启jmx服务
java可以利用编写mbeans来暴露很多程序指标,jdk自己也有很多指标,例如内存使用以及划分.很多小伙伴都表示这些值不少jstat就可以获取查看的吗,ssh登录上去就可以,也很方便.jstat确实 ...
【kafka】JMX 监控kafka FINER RMI TCP getConnectionId IOException
1.背景使用Jmx监控kafka相关信息,但是运行的时候报错如下我的代码大致逻辑是 JMXServiceUrl jmx = new JMXServiceUrl(url) JMXConnector ...
【kafka】JMX 监控kafka kafka rmi NoSuchObjectException no such object in table
1.背景使用Jmx监控kafka相关信息,但是运行的时候报错如下我的代码大致逻辑是 JMXServiceUrl jmx = new JMXServiceUrl(url) JMXConnector ...
【kafka】Kafka常用JMX监控指标整理
文章目录 1.概述一.系统相关指标二.GC相关指标三.JVM相关指标四.Topic相关指标五.Broker相关指标 6.kafka生产者指标 7.Kafka消费者指标 7.1 0.8.2.2 ...

3. kafka开启JMX

开启JMX

jconsole连接

程序获取

3. kafka开启JMX相关推荐

最新文章

热门文章