开启JMX
kafka开启JMX的2种方式：
- 启动kafka时增加JMX_PORT=9988，即
JMX_PORT=9988 bin/kafka-server-start.sh -daemon config/server.properties - 修改
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 ]; then
KAFKA_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" ]; then
KAFKA_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 ]; then
KAFKA_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 {
// 初始化连接jmx
JMXServiceURL 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无关的metric
Object o1 = jmxConn.getValue(
"kafka.server:type=ReplicaManager,name=PartitionCount",
Lists.newArrayList("Value"));
System.out.println(o1);
// 与topic有关的metric
Object o2 = jmxConn.getValue(topicName,
"kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec",
Lists.newArrayList("Count", "OneMinuteRate", "FiveMinuteRate"));
System.out.println(o2);
Thread.sleep(5000);
}
}
}
