通过TelnetClient获取Zookeeper监控数据

387次阅读

共计 7071 个字符，预计需要花费 18 分钟才能阅读完成。

如果想编写一个监控 Zookeeper 的 Java 程序，可以通过两种方式 :

（1）通过 TelnetClient 发送命令 , 命令的详解参考：http://zookeeper.apache.org/doc/trunk/zookeeperAdmin.html#sc_zkCommands

（2）通过 JMX，说明请参考：http://zookeeper.apache.org/doc/trunk/zookeeperJMX.html

本文通过一个简单的例子来演示如何通过 TelnetClient 发送 mntr 命令获取 Zookeeper 的监控数据

写一个 Telnet 的工具类

package com.eric.agent.utils;
import org.apache.commons.net.telnet.TelnetClient;
import java.io.IOException;
import java.io.InputStream;
import java.io.PrintStream;
/**
* Telnet 操作器, 基于 commons-net-2.2.jar
*
* @author aihua.sun
* @date 2015/4/9
* @since V1.0
*/
public class TelnetTools {
private String prompt = “>”; // 结束标识字符串,Windows 中是 >,Linux 中是 #
private char promptChar = ‘>’; // 结束标识字符
private TelnetClient telnet;
private InputStream in; // 输入流, 接收返回信息
private PrintStream out; // 向服务器写入命令
/**
* @param termtype 协议类型：VT100、VT52、VT220、VTNT、ANSI
* @param prompt 结果结束标识
*/
public TelnetTools(String termtype, String prompt) {
telnet = new TelnetClient(termtype);
setPrompt(prompt);
}
public TelnetTools(String termtype) {
telnet = new TelnetClient(termtype);
}
public TelnetTools() {
telnet = new TelnetClient();
}
/**
* 登录到目标主机
*
* @param ip
* @param port
*/
public void login(String ip, int port) {
try {
telnet.connect(ip, port);
in = telnet.getInputStream();
out = new PrintStream(telnet.getOutputStream());
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* 读取分析结果
*
* @param pattern 匹配到该字符串时返回结果
* @return
*/
public String readUntil(String pattern) {
StringBuffer sb = new StringBuffer();
try {
char lastChar = (char) -1;
boolean flag = pattern != null && pattern.length() > 0;
if (flag)
lastChar = pattern.charAt(pattern.length() – 1);
char ch;
int code = -1;
while ((code = in.read()) != -1) {
ch = (char) code;
sb.append(ch);
// 匹配到结束标识时返回结果
if (flag) {
if (ch == lastChar && sb.toString().endsWith(pattern)) {
return sb.toString();
}
} else {
// 如果没指定结束标识, 匹配到默认结束标识字符时返回结果
if (ch == promptChar)
return sb.toString();
}
// 登录失败时返回结果
if (sb.toString().contains(“Login Failed”)) {
return sb.toString();
}
}
} catch (Exception e) {
e.printStackTrace();
}
return sb.toString();
}
/**
* 发送命令
*
* @param value
*/
public void write(String value) {
try {
out.println(value);
out.flush();
} catch (Exception e) {
e.printStackTrace();
}
}
/**
* 发送命令, 返回执行结果
*
* @param command
* @return
*/
public String sendCommand(String command) {
try {
write(command);
return readUntil(prompt);
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
/**
* 关闭连接
*/
public void distinct() {
try {
if (telnet != null && !telnet.isConnected())
telnet.disconnect();
} catch (IOException e) {
e.printStackTrace();
}
}
public void setPrompt(String prompt) {
if (prompt != null) {
this.prompt = prompt;
this.promptChar = prompt.charAt(prompt.length() – 1);
}
}
}

调用类

package com.tscloud.agent.flume.source.dataprovider;

/**
* 通过 HTTP 作为 HDFS Master 监控信息的 source
* 明细可参考 http://zookeeper.apache.org/doc/r3.4.6/zookeeperAdmin.html#sc_zkCommands
* @author aihua.sun
* @date 2015/4/6
* @since V1.0
*/

import com.eric.agent.flume.model.ZookeeperRoleInfo;
import com.eric.agent.flume.source.base.IClusterServiceRoleDataProvider;
import com.eric.agent.utils.AgentConstants;
import com.eric.agent.utils.TelnetTools;
import com.eric.common.utils.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.UnsupportedEncodingException;
import java.util.*;

public class ZookeeperDataProvider{
protected final Logger LOGGER = LoggerFactory.getLogger(getClass());
private static final String zk_avg_latency = “zk_avg_latency”;
private static final String zk_max_latency = “zk_max_latency”;
private static final String zk_min_latency = “zk_min_latency”;
private static final String zk_packets_received = “zk_packets_received”;
private static final String zk_packets_sent = “zk_packets_sent”;
private static final String zk_server_state = “zk_server_state”;
private static final String zk_znode_count = “zk_znode_count”;
private static final String zk_followers = “zk_followers”;
private static final String zk_open_file_descriptor_count = “zk_open_file_descriptor_count”;

public String extractMonitorData() {
//TODO 通过调用 API 获得 IP 以及参数
ZookeeperRoleInfo monitorDataPoint = new ZookeeperRoleInfo();
String IP = “192.168.40.242”;
int port = 2181;
TelnetTools telnet=null;
try {
telnet = new TelnetTools();
telnet.login(IP, port);
String rs = telnet.sendCommand(“mntr”);
Map<String, String> telnetResultMap = parseTelnetResult(rs);
monitorDataPoint.setZkAvgLatency(translateStrToLong(telnetResultMap.get(zk_avg_latency)));
monitorDataPoint.setZkMaxLatency(translateStrToLong(telnetResultMap.get(zk_max_latency)));
monitorDataPoint.setZkMinLatency(translateStrToLong(telnetResultMap.get(zk_min_latency)));
monitorDataPoint.setZkPacketsReceived(translateStrToLong(telnetResultMap.get(zk_packets_received)));
monitorDataPoint.setZkPacketsSent(translateStrToLong(telnetResultMap.get(zk_packets_sent)));
monitorDataPoint.setZkServerState(telnetResultMap.get(zk_server_state));
monitorDataPoint.setZkZnodeCount(translateStrToLong(telnetResultMap.get(zk_znode_count)));
monitorDataPoint.setZkFollowers(translateStrToLong(telnetResultMap.get(zk_followers)));
monitorDataPoint.setZkOpenFileDescriptorCount(translateStrToLong(telnetResultMap.get(zk_open_file_descriptor_count)));
} catch (Exception e) {
e.printStackTrace();
}finally{
telnet.distinct();
}

return monitorDataPoint.toString();
}

private Long translateStrToLong(String value) {
if (org.apache.commons.lang.StringUtils.isAlphanumeric(value)) {
return Long.valueOf(value);
}
return 0L;
}

private Map<String, String> parseTelnetResult(String rs) {
//The output contains multiple lines with the following format:
//key \t value
String[] resultArray = rs.split(“\n”);
Map<String, String> resultMap = new HashMap<String, String>();
for (String recordLine : resultArray) {
String[] recordKeyValue = recordLine.split(“\t”);
LOGGER.debug(“############recordKeyValue.size:” + recordKeyValue.length + ” recordKeyValue:” + Arrays.toString(recordKeyValue));
if (recordKeyValue != null && recordKeyValue.length == 2) {
resultMap.put(recordKeyValue[0], recordKeyValue[1]);
}
}
return resultMap;
}

public static void main(String[] args) {
System.out.println(new ZookeeperDataProvider().extractMonitorData());
}

}

————————————– 分割线 ————————————–

Ubuntu 14.04 安装分布式存储 Sheepdog+ZooKeeper http://www.linuxidc.com/Linux/2014-12/110352.htm

CentOS 6 安装 sheepdog 虚拟机分布式储存 http://www.linuxidc.com/Linux/2013-08/89109.htm

ZooKeeper 集群配置 http://www.linuxidc.com/Linux/2013-06/86348.htm

使用 ZooKeeper 实现分布式共享锁 http://www.linuxidc.com/Linux/2013-06/85550.htm

分布式服务框架 ZooKeeper — 管理分布式环境中的数据 http://www.linuxidc.com/Linux/2013-06/85549.htm

ZooKeeper 集群环境搭建实践 http://www.linuxidc.com/Linux/2013-04/83562.htm

ZooKeeper 服务器集群环境配置实测 http://www.linuxidc.com/Linux/2013-04/83559.htm

ZooKeeper 集群安装 http://www.linuxidc.com/Linux/2012-10/72906.htm

本文永久更新链接地址 ：http://www.linuxidc.com/Linux/2015-04/116178.htm