心跳检测的思路及代码

Posted 2020-10-19 星朝

外网服务端保存内网服务端会话的有效性以及平台上监控所有内网服务端的网络状况，模仿心跳机制实现，这里在做一点叙诉，关于思路和具体实现。

在很多的平台应用中，都有这样的需求，平台内包括多个子系统或者属于其管控范围内的其他平台，需要对这些系统进行统一的监控，来查看当前的运行状态或者其他运行信息，我们的应用也有这样的一个情况，需要再外网服务端（平台）上监控，其下运行的多个内网服务端的网络状况，查阅了写资料后确立了2种可实现的方式。
1：轮询机制

2：心跳机制

先简单介绍一下，
轮询：概括来说是服务端定时主动的去与要监控状态的客户端（或者叫其他系统）通信，询问当前的某种状态，客户端返回状态信息，客户端没有返回或返回错误、失效信息、则认为客户端已经宕机，然后服务端自己内部把这个客户端的状态保存下来（宕机或者其他），如果客户端正常，那么返回正常状态，如果客户端宕机或者返回的是定义的失效状态那么当前的客户端状态是能够及时的监控到的，如果客户端宕机之后重启了那么当服务端定时来轮询的时候，还是可以正常的获取返回信息，把其状态重新更新。

心跳：最终得到的结果是与轮询一样的但是实现的方式有差别，心跳不是服务端主动去发信息检测客户端状态，而是在服务端保存下来所有客户端的状态信息，然后等待客户端定时来访问服务端，更新自己的当前状态，如果客户端超过指定的时间没有来更新状态，则认为客户端已经宕机或者其状态异常。

心跳机制与轮询的比较，在我们的应用中，采用的是心跳，这样一是避免服务端的压力，二是灵活好控制，上一篇文章中提到过，我们的外网服务端（服务端）不知道内网服务端（客户端）的地址，有虽然有保存客户端的socket会话，但是客户端宕机会话就失效了。所以只能等着他主动来报告状态。

在来说一下实现方式，这个很简单，就是一个思路问题。

首先，客户端（内网服务端）启动后，带着自己的标识符与服务端建立socket连接，服务端缓存下来对应信息（上一篇文章中已经实现过了），然后在通过socket流，定时发送当前信息消息到服务端（外网服务器端）某个接口，服务端收到后更新当前的客户端的状态，比如（会话地址，标识符，网络的活跃状态，连接时间，心跳时间），本次来更新的时间就是心跳时间，然后服务端还有一个定时器，定时检查所有缓存的客户端会话集合，将其中心跳时间与当前时间进行对比，如果超过指定的时间还没有来更新则认为该客户端的网络出现异常或者宕机，然后更新该客户端的网络状态。

[java] view plain copy

1. import java.io.BufferedReader;  
2. import java.io.IOException;  
3. import java.io.InputStream;  
4. import java.io.InputStreamReader;  
5. import java.net.InetSocketAddress;  
6. import java.net.ServerSocket;  
7. import java.net.Socket;  
8. import java.util.Date;  
9. import java.util.HashMap;  
10. import java.util.Map;  
11.   
12. import org.dom4j.Document;  
13. import org.dom4j.DocumentException;  
14. import org.dom4j.DocumentHelper;  
15.   
16. import cn.edu.zju.cst.mina.im.server.entity.User;  
17. import cn.edu.zju.cst.mina.im.server.handler.ServerControler;  
18.   
19. public class UserStateManage extends Thread {  
20.       
21.     //在线用户状态列表  
22.     static HashMap<Integer, UserState> userStateList = new HashMap<Integer, UserState>();  
23.     Object hashLock = new Object();  
24.       
25.     //当前的连接数和工作线程数  
26.     static int workThreadNum = 0;  
27.     static int socketConnect = 0;  
28.       
29.     private ServerSocket serverSocket;  
30.     //服务器IP  
31.     private String host = "10.82.81.79";  
32.       
33.     //服务器端口  
34.     private int stateReportPort = 60001;  
35.       
36.     //设置心跳包的结束标记  
37.     String endFlag = "</protocol>";  
38.     CharSequence csEndFlag = endFlag.subSequence(0, 10);  
39.       
40.     //扫描间隔  
41.     private int scanTime = 1800;   
42.       
43.       
44.     @Override  
45.     public void run() {  
46.         //绑定端口,并开始侦听用户的心跳包  
47.         serverSocket = startListenUserReport(stateReportPort);  
48.         if(serverSocket == null){  
49.             System.out.println("【创建ServerSocket失败！】");  
50.             return;  
51.         }  
52.         //启动扫描线程  
53.         Thread scanThread = new Thread(new scan());  
54.         scanThread.start();  
55.         //等待用户心跳包请求  
56.         while(true){  
57.             Socket socket = null;  
58.             try {  
59.                 socketConnect = socketConnect + 1;  
60.                 //接收客户端的连接  
61.                 socket = serverSocket.accept();     
62.                 //为该连接创建一个工作线程  
63.                 Thread workThread = new Thread(new Handler(socket));  
64.                 //启动工作线程  
65.                 workThread.start();  
66.             } catch (IOException e) {  
67.                 e.printStackTrace();  
68.             }  
69.         }  
70.     }  
71.       
72.     /** 
73.      * 创建一个ServerSocket来侦听用户心跳包请求 
74.      * @param port 指定的服务器端的端口 
75.      * @return 返回ServerSocket 
76.      * @author dream 
77.     */  
78.     public ServerSocket startListenUserReport(int port){  
79.         try {  
80.             ServerSocket serverSocket = new ServerSocket();  
81.             if(!serverSocket.getReuseAddress()){  
82.                 serverSocket.setReuseAddress(true);  
83.             }  
84.             serverSocket.bind(new InetSocketAddress(host,port));  
85.             System.out.println("【开始在"+serverSocket.getLocalSocketAddress()+"上侦听用户的心跳包请求！】");  
86.             return serverSocket;  
87.         } catch (IOException e) {  
88.             System.out.println("【端口"+port+"已经被占用！】");  
89.             if (serverSocket != null) {  
90.                 if (!serverSocket.isClosed()) {  
91.                     try {  
92.                         serverSocket.close();  
93.                     } catch (IOException e1) {  
94.                         e1.printStackTrace();  
95.                     }  
96.                 }  
97.             }  
98.         }  
99.         return serverSocket;  
100.     }  
101.       
102.       
103.     //工作线程类  
104.     class Handler implements Runnable{  
105.         private Socket socket;  
106.         UserState us = null;  
107.         User newUser = null;  
108.         private int userId;  
109.         private int userState;  
110.         /** 
111.          * 构造函数，从调用者那里取得socket 
112.          * @param socket 指定的socket 
113.          * @author dream 
114.         */  
115.         public Handler(Socket socket){  
116.             this.socket = socket;  
117.         }  
118.           
119.         /** 
120.          * 从指定的socket中得到输入流 
121.          * @param socket 指定的socket 
122.          * @return 返回BufferedReader 
123.          * @author dream 
124.          */  
125.         private BufferedReader getReader(Socket socket){  
126.             InputStream is = null;  
127.             BufferedReader br = null;  
128.   
129.             try {  
130.                 is = socket.getInputStream();  
131.                 br = new BufferedReader(new InputStreamReader(is));  
132.             } catch (IOException e) {  
133.                 e.printStackTrace();  
134.             }  
135.             return br;  
136.         }  
137.           
138.         public void run() {  
139.             try{  
140.                 workThreadNum = workThreadNum +1;  
141.                 System.out.println("【第"+workThreadNum+"个的连接:"+socket.getInetAddress()+":"+socket.getPort()+"】");  
142.                 BufferedReader br = getReader(socket);  
143.                 String meg = null;  
144.                 StringBuffer report = new StringBuffer();  
145.                 while ((meg = br.readLine()) != null) {  
146.                     report.append(meg);  
147.                     if (meg.contains(csEndFlag)) {  
148.                         us = getReporterUserState(meg, socket);  
149.                         synchronized (hashLock) {  
150.                             userStateList.put(userId, us);  
151.                         }  
152.                     }  
153.                 }  
154.             }catch(IOException e){  
155.                 System.out.println("【客户:"+newUser.getUser_id()+"已经断开连接！】");  
156.                 userStateList.remove( userId );  
157.                 announceStateChange( userId , -1);  
158.             }finally{  
159.                 if(socket != null){  
160.                     try {  
161.                         //断开连接  
162.                         socket.close();  
163.                     } catch (IOException e) {  
164.                         e.printStackTrace();  
165.                     }  
166.                 }  
167.             }  
168.         }  
169.         private UserState getReporterUserState(String meg , Socket socket){  
170.             UserState us = new UserState();  
171.             try {  
172.                 Document requestDoc = DocumentHelper.parseText(meg);  
173.                 newUser = ServerControler.parseXmlToUserState(requestDoc,socket);  
174.                 userId = newUser.getUser_id();  
175.                 userState = newUser.getUser_state();  
176.                 us.setFlag(2);  
177.                 us.setUser_state( userState );  
178.                 us.setUser_id( userId );  
179.                 us.setUser_ip(newUser.getUser_ip());  
180.                 us.setUser_port(newUser.getUser_port());  
181.             } catch (DocumentException e) {  
182.                 System.out.println("【来自客户端的信息不是一个合法的心跳包协议】");  
183.             }  
184.             return us;  
185.         }  
186.     }  
187.       
188.     //扫描线程  
189.     class scan implements Runnable{  
190.         public void run() {  
191.             while (true) {  
192.                 System.out.println("*******"+new Date()+"：扫描线程开始扫描"+"*******");  
193.                 synchronized (hashLock) {  
194.                     if(!userStateList.isEmpty()){  
195.                         //遍历在线用户列表  
196.                         for (Map.Entry<Integer, UserState> entry : userStateList.entrySet()) {  
197.                             int flag = entry.getValue().getFlag();  
198.                             if ( (flag - 1) < 0) {  
199.                                 //在这里通知该用户的好友其状态发生改变  
200. //                              announceStateChange(entry.getKey() , 0);  
201.                             }else{  
202.                                 entry.getValue().setFlag(flag - 1);  
203.                                 userStateList.put(entry.getKey(), entry.getValue());  
204.                             }  
205.                             System.out.println(entry.getKey() + "-->" + entry.getValue().toString());  
206.                         }  
207.                     }else{  
208.                         System.out.println("现在还没有在线用户！");   
209.                     }  
210.                 }  
211.                 //实现定时扫描  
212.                 try {  
213.                     sleep(scanTime);  
214.                 } catch (InterruptedException e) {  
215.                     e.printStackTrace();  
216.                 }  
217.             }  
218.         }  
219.     }  
220.       
221.     private void announceStateChange(int userId , int state){  
222.         System.out.println("通知其好友!");  
223.     }  
224.       
225.     /** 
226.      * 查询一个用户是否在线 
227.      * @param userId 指定要查询状态的用户的ID 
228.      * @return true 在线； false 不在线； 
229.      * @author dream 
230.     */  
231.     public boolean isAlive(int userId){  
232.         synchronized (hashLock) {  
233.             return userStateList.containsKey(userId);  
234.         }  
235.     }  
236.       
237.     /** 
238.      * 返回指定用户ID的状态 
239.      * @param userId 指定要查询状态的用户的ID 
240.      * @return >0 该用户在线;  -1 该用户离线 
241.      * @author dream 
242.     */  
243.     public int getUserState(int userId){  
244.         synchronized (hashLock) {  
245.             if(userStateList.containsKey(userId)){  
246.                 return userStateList.get(userId).getUser_state();  
247.             }else{  
248.                 return -1;  
249.             }  
250.         }  
251.     }  
252.       
253.     public Object getHashLock() {  
254.         return hashLock;  
255.     }  
256.   
257.     public void setHashLock(Object hashLock) {  
258.         this.hashLock = hashLock;  
259.     }  
260.   
261.     public String getHost() {  
262.         return host;  
263.     }  
264.   
265.     public void setHost(String host) {  
266.         this.host = host;  
267.     }  
268.   
269.     public int getStateReportPort() {  
270.         return stateReportPort;  
271.     }  
272.   
273.     public void setStateReportPort(int stateReportPort) {  
274.         this.stateReportPort = stateReportPort;  
275.     }  
276.   
277.     public String getEndFlag() {  
278.         return endFlag;  
279.     }  
280.   
281.     public void setEndFlag(String endFlag) {  
282.         this.endFlag = endFlag;  
283.     }  
284.   
285.     public int getScanTime() {  
286.         return scanTime;  
287.     }  
288.   
289.     public void setScanTime(int scanTime) {  
290.         this.scanTime = scanTime;  
291.     }  
292.   
293.     public static HashMap<Integer, UserState> getUserStateList() {  
294.         return userStateList;  
295.     }  
296.   
297.     public static int getWorkThreadNum() {  
298.         return workThreadNum;  
299.     }  
300.   
301.     public static int getSocketConnect() {  
302.         return socketConnect;  
303.     }  
304.     //测试本函数的main函数  
305.     public static void main(String arg[]){  
306.         UserStateManage usm = new UserStateManage();  
307.         usm.start();  
308.     }  
309. }