摘要: 原创出处 http://www.iocoder.cn/Elastic-Job/job-monitor/ 「芋道源码」欢迎转载,保留摘要,谢谢!
本文基于 Elastic-Job V2.1.5 版本分享
本文主要分享 Elastic-Job-Lite 作业监控服务。内容对应《官方文档 —— DUMP作业运行信息》。
使用Elastic-Job-Lite过程中可能会碰到一些分布式问题,导致作业运行不稳定。 由于无法在生产环境调试,通过dump命令可以把作业内部相关信息dump出来,方便开发者debug分析; 另外为了不泄露隐私,已将相关信息中的ip地址以ip1, ip2…的形式过滤,可以在互联网上公开传输环境信息,便于进一步完善Elastic-Job。
涉及到主要类的类图如下( 打开大图 ):
你行好事会因为得到赞赏而愉悦 同理,开源项目贡献者会因为 Star 而更加有动力 为 Elastic-Job 点赞!传送门
MonitorService,作业监控服务。
初始化 MonitorService 方法实现如下:
// MonitorService.java
private final String jobName;
public void listen() {
int port = configService.load(true).getMonitorPort();
if (port < 0) {
return;
}
try {
log.info("Elastic job: Monitor service is running, the port is '{}'", port);
openSocketForMonitor(port);
} catch (final IOException ex) {
log.error("Elastic job: Monitor service listen failure, error is: ", ex);
}
}
private void openSocketForMonitor(final int port) throws IOException {
serverSocket = new ServerSocket(port);
new Thread() {
@Override
public void run() {
while (!closed) {
try {
process(serverSocket.accept());
} catch (final IOException ex) {
log.error("Elastic job: Monitor service open socket for monitor failure, error is: ", ex);
}
}
}
}.start();
}
LiteJobConfiguration.monitorPort
)启动 ServerSocket。一个作业对应一个作业监控端口,所以配置时,请不要重复端口噢。处理 dump命令 方法如下:
// MonitorService.java
private void process(final Socket socket) throws IOException {
try (
BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
BufferedWriter writer = new BufferedWriter(new OutputStreamWriter(socket.getOutputStream()));
Socket autoCloseSocket = socket) {
// 读取命令
String cmdLine = reader.readLine();
if (null != cmdLine && DUMP_COMMAND.equalsIgnoreCase(cmdLine)) { // DUMP
List<String> result = new ArrayList<>();
dumpDirectly("/" + jobName, result);
outputMessage(writer, Joiner.on("\n").join(SensitiveInfoUtils.filterSensitiveIps(result)) + "\n");
}
}
}
#process()
方法,目前只支持 DUMP
命令。如果你有自定义命令的需要,可以拓展该方法。#dumpDirectly()
方法,输出当前作业名对应的相关调试信息。
private void dumpDirectly(final String path, final List<String> result) { for (String each : regCenter.getChildrenKeys(path)) { String zkPath = path + "/" + each; String zkValue = regCenter.get(zkPath); if (null == zkValue) { zkValue = ""; } TreeCache treeCache = (TreeCache) regCenter.getRawCache("/" + jobName); ChildData treeCacheData = treeCache.getCurrentData(zkPath); String treeCachePath = null == treeCacheData ? "" : treeCacheData.getPath(); String treeCacheValue = null == treeCacheData ? "" : new String(treeCacheData.getData()); // 判断 TreeCache缓存 和 注册中心 数据一致 if (zkValue.equals(treeCacheValue) && zkPath.equals(treeCachePath)) { result.add(Joiner.on(" | ").join(zkPath, zkValue)); } else { result.add(Joiner.on(" | ").join(zkPath, zkValue, treeCachePath, treeCacheValue)); } // 递归 dumpDirectly(zkPath, result); } }