监控心跳实现

监控主动轮询进程是否会崩溃跪掉，采取措施是通过心跳包形式进行抓取，定时生成文件，jnotify监听发到kafka上，之后通过消费者进行解析，若发现不符规则情况，发邮件短信报警。

Java代码

1. package heartbeat.monitor;
2. import java.io.FileInputStream;
3. import java.io.IOException;
4. import java.io.InputStream;
5. import java.util.ArrayList;
6. import java.util.List;
7. import java.util.Properties;
8. import java.util.concurrent.ExecutionException;
9. import org.apache.commons.io.IOUtils;
10. import heartbeat.monitor.sendmessage.SendMailUtil;
11. import heartbeat.monitor.util.HeartBeatMonitorConstant;
12. /**
13. * @author Switching
14. * @version 1.0, 2017-01-01
15. * @since heartbeat.monitor 1.0
16. */
17. public class HeartBeatMonitorAccess {
18. private List<HeartBeatMonitor> consumers;
19. private static String heartbeatConsumerProperties;
20. public HeartBeatMonitorAccess() throws IOException {
21. Properties properties = new Properties();
22. if (heartbeatConsumerProperties == null) {
23. properties.load(ClassLoader.getSystemResourceAsStream("heartBeatMonitor.properties"));
24. } else {
25. String propurl = heartbeatConsumerProperties;
26. InputStream in = new FileInputStream(propurl);
27. properties.load(in);
28. IOUtils.closeQuietly(in);
29. }
30. int num = 1;
31. String topic = properties.getProperty(HeartBeatMonitorConstant.MONITOR_TOPIC);
32. consumers = new ArrayList<HeartBeatMonitor>(num);
33. for (int i = 0; i < num; i++) {
34. consumers.add(new HeartBeatMonitor(properties, topic));
35. new SendMailUtil(properties, topic);
36. }
37. }
38. public void exeute() throws InterruptedException, ExecutionException {
39. for (HeartBeatMonitor consumer : consumers) {
40. new Thread(consumer).start();
41. }
42. }
43. public static void main(String[] args) throws Exception {
44. if (args.length > 0)
45. heartbeatConsumerProperties = args[0];
46. HeartBeatMonitorAccess consumerGroup = new HeartBeatMonitorAccess();
47. consumerGroup.exeute();
48. }
49. }

消费者抓取相应的规则进行解析判断 
后台轮询消费kafka中消息

Java代码

1. package heartbeat.monitor;
2. import java.io.IOException;
3. import java.text.DateFormat;
4. import java.text.ParseException;
5. import java.text.SimpleDateFormat;
6. import java.util.Collections;
7. import java.util.Date;
8. import java.util.Map;
9. import java.util.Properties;
10. import org.apache.kafka.clients.consumer.ConsumerRecord;
11. import org.apache.kafka.clients.consumer.ConsumerRecords;
12. import org.apache.kafka.clients.consumer.KafkaConsumer;
13. import org.codehaus.jackson.map.ObjectMapper;
14. import heartbeat.monitor.sendmessage.SendMailUtil;
15. import heartbeat.monitor.util.HeartBeatMonitorConstant;
16. import heartbeat.monitor.util.MessageUtil;
17. /**
18. * @author Switching
19. * @version 1.0, 2017-01-01
20. * @since heartbeat.monitor 1.0
21. */
22. public class HeartBeatMonitor implements Runnable {
23. private final KafkaConsumer<String, String> consumer;
24. public static String lastMonitorTime;
25. public static String lastMonitorInfo;
26. public static DateFormat dfMonitorTime = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
27. private long timeMonitor = 20000;
28. private long maxTriesTime = 3;
29. private String monitorKeyWord = "FILE_WRITTEN_CLOSED";
30. private String monitorLogLvls = "EXCEPTION";
31. public HeartBeatMonitor(Properties properties, String topic) throws IOException {
32. if (!MessageUtil.isEmpty(properties.getProperty(HeartBeatMonitorConstant.MONITOR_OFFSET_TIME)))
33. timeMonitor = Long.parseLong(properties.getProperty(HeartBeatMonitorConstant.MONITOR_OFFSET_TIME));
34. if (!MessageUtil.isEmpty(properties.getProperty(HeartBeatMonitorConstant.MONITOR_TRIES)))
35. maxTriesTime = Long.parseLong(properties.getProperty(HeartBeatMonitorConstant.MONITOR_TRIES));
36. if (!MessageUtil.isEmpty(properties.getProperty(HeartBeatMonitorConstant.MONITOR_KEY_WORD)))
37. monitorKeyWord = properties.getProperty(HeartBeatMonitorConstant.MONITOR_KEY_WORD);
38. if (!MessageUtil.isEmpty(properties.getProperty(HeartBeatMonitorConstant.MONITOR_LOG_LVLS)))
39. monitorLogLvls = properties.getProperty(HeartBeatMonitorConstant.MONITOR_LOG_LVLS);
40. consumer = new KafkaConsumer<String, String>(properties);
41. consumer.subscribe(Collections.singletonList(topic));
42. }
43. public void close() {
44. consumer.close();
45. }
46. public void run() {
47. long triesTime = maxTriesTime - 1;
48. System.out.println("Heartbeat Monitor Start!");
49. if (lastMonitorTime == null) {
50. resetTime();
51. }
52. String name = Thread.currentThread().getName();
53. while (true) {
54. Date dtMonitor;
55. Date dtLocal = new Date();
56. ConsumerRecords<String, String> records = consumer.poll(2);
57. for (ConsumerRecord<String, String> record : records) {
58. ObjectMapper objectMapper = new ObjectMapper();
59. try {
60. Map<String, String> recordMap = objectMapper.readValue(record.value(), Map.class);
61. if (recordMap.get(HeartBeatMonitorConstant.FILE_EVENT).toString().equals(monitorKeyWord)) {
62. lastMonitorTime = recordMap.get(HeartBeatMonitorConstant.FILE_TIME).toString();
63. lastMonitorInfo = recordMap.get(HeartBeatMonitorConstant.SFTP_HOST_NAME) + ":"
64. + recordMap.get(HeartBeatMonitorConstant.SFTP_HOST_IP);
65. }
66. } catch (Exception e) {
67. e.printStackTrace();
68. }
69. System.out.println(name + "---" + record.partition() + ":" + record.offset() + " = " + record.key()
70. + ":" + record.value());
71. }
72. try {
73. dtMonitor = dfMonitorTime.parse(lastMonitorTime);
74. if ((dtLocal.getTime() - timeMonitor) > dtMonitor.getTime()) {
75. if (triesTime > 0) {
76. triesTime--;
77. } else {
78. resetTime();
79. triesTime = maxTriesTime - 1;
80. }
81. if (HeartBeatMonitorConstant.LOG_LVL_EXCEP.equals(monitorLogLvls)
82. || HeartBeatMonitorConstant.LOG_LVL_ALL.equals(monitorLogLvls)) {
83. String content = "[" + dtLocal + "] " + lastMonitorInfo
84. + " Exception! manual intervention please! " + (triesTime + 1) + " times! \n\t\t\t\tNow("
85. + dtLocal + ") is beyond " + timeMonitor / 1000 + "(s) lastMonitorTime(" + dtMonitor
86. + ").";
87. System.out.println(content);
88. if (triesTime == 0) {
89. SendMailUtil.sendMailAccess(lastMonitorInfo, content);
90. }
91. }
92. } else {
93. if (HeartBeatMonitorConstant.LOG_LVL_NOEXCEP.equals(monitorLogLvls)
94. || HeartBeatMonitorConstant.LOG_LVL_ALL.equals(monitorLogLvls)) {
95. System.out.println("[" + dtLocal + "] " + name + "NoException! \n\t\t\t\tNow(" + dtLocal
96. + ") is Range " + timeMonitor / 1000 + " (s) lastMonitorTime(" + dtMonitor + ").");
97. }
98. }
99. } catch (ParseException e) {
100. e.printStackTrace();
101. } catch (Exception e) {
102. e.printStackTrace();
103. }
104. }
105. }
106. public void resetTime() {
107. lastMonitorTime = dfMonitorTime.format(new Date());
108. if (lastMonitorInfo == null) {
109. lastMonitorInfo = "Monitor";
110. }
111. }
112. }

通过配置文件形式，进行各种事件的监控

Java代码

1. #Created by Switching
2. auto.commit.interval.ms=1000
3. auto.offset.reset=earliest
4. bootstrap.servers=192.168.102.10\:9092
5. enable.auto.commit=true
6. group.id=monitor1
7. key.deserializer=org.apache.kafka.common.serialization.StringDeserializer
8. monitorTopic=monitor
9. value.deserializer=org.apache.kafka.common.serialization.StringDeserializer
10. #heartBeat monitor exception notify times(3)
11. monitorTries=3
12. #heartBeat monitor offset time(20000)(ms)
13. monitorOffsetTime=20000
14. #heartBeat monitor log lvls(EXCEPTION):ALL|EXCEPTION|NOEXCEPTION|NONE
15. monitorLogLvls=EXCEPTION
16. #heartBeat Monitor mailInfo
17. sendEmailAccount=xxxx@jdongtech.com
18. sendEmailPassword=xxxxx
19. sendEmailSMTPHost=smtp.xxxx.com
20. receEMailAccount=xxx@jdongtech.com