第九篇 - 手写心跳检测

Github源码下载：https://github.com/chenxingxing6/sourcecode/tree/master/study-net

一、前言

1.1 什么是心跳检测

在分布式系统中，分布在不同主机上的节点需要检测其他节点的状态，如服务器节点需要检测从节点是否失效。为了检测对方节点的有效性，每隔固定时间就发送一个固定信息给对方，对方回复一个固定信息，如果长时间没有收到对方的回复，则断开与对方的连接。发包方既可以是服务端，也可以是客户端。因为是每隔固定时间发送一次，类似心跳，所以发送的固定信息称为心跳包。心跳包一般为比较小的包，可根据具体实现。一般而言，应该客户端主动向服务器发送心跳包，因为服务器向客户端发送心跳包会影响服务器的性能。

所有保持长连接的地方都要用到心跳包，心跳包就是在客户端和服务器间定时通知对方自己状态的一个自己定义的命令字，按照一定的时间间隔发送，类似于心跳，所以叫做心跳包。

1.2 实现方式

二、自己在应用层实现心跳检测

客户端：

1.Client通过持有Socket的对象，可以发送Massage Object(消息)给服务端。
2.如果keepAliveDelay 2秒内未发送任何数据，则自动发送一个KeepAlive(心跳)给服务端，用于维持连接。

服务端：

1.由于客户端会定时（keepAliveDelay毫秒）发送维持连接的信息过来，所以，服务端要有一个检测机制。
2.当服务端receiveTimeDelay毫秒（程序中是3秒）内未接收任何数据，则自动断开与客户端的连接。

2.1 Client.java

package com.demo.longconn.client;
import com.demo.longconn.KeepAlive;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.util.concurrent.TimeUnit;/*** User: lanxinghua* Date: 2019/11/4 18:49* Desc: 客户端，定时向服务端程序，发送一个维持连接包的*/
public class Client {private String ip;private int port;private String name;private Socket socket;// 连接状态private volatile boolean isConn = false;// 最后一次发送数据时间private long lastSendTime;public Client(String ip, int port, String name) {this.ip = ip;this.port = port;this.name = name;}public void start(){if (isConn)return;try {System.out.println(name + "已启动.....");socket = new Socket(ip, port);isConn = true;// 保持长连接的线程，每隔2秒项服务器发一个一个保持连接的心跳消息lastSendTime = System.currentTimeMillis();new Thread(new KeepAliveWatchDog()).start();// 接受消息的线程，处理消息// new Thread(new ReceiveWatchDog()).start();}catch (Exception e){e.printStackTrace();stop();}}public void stop(){if (isConn){isConn = false;}}class KeepAliveWatchDog implements Runnable{// 连接推迟时间2slong keepAliveDelay = 2000;// 检测推迟时间long checkDelay = 10;public void run() {if (isConn == false){System.out.println(isConn);}while (isConn){long tt = (System.currentTimeMillis() - lastSendTime);// 到了时间需要去发送心跳检测if (tt > keepAliveDelay){try {ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());oos.writeObject(new KeepAlive(name));oos.flush();lastSendTime = System.currentTimeMillis();// 假设客户端宕机if ("客户端3".equals(name)){System.out.println(name + "出现故障，将在1s后宕机");TimeUnit.SECONDS.sleep(1);stop();}if ("客户端2".equals(name)){System.out.println(name + "出现故障，将在2s后宕机");TimeUnit.SECONDS.sleep(2);stop();}}catch (Exception e){e.printStackTrace();stop();}}else {try {TimeUnit.MILLISECONDS.sleep(checkDelay);}catch (Exception e){e.printStackTrace();stop();}}}}}class ReceiveWatchDog implements Runnable{public void run() {while (isConn){try {InputStream in = socket.getInputStream();// 判断流里面的字节数if (in.available() > 0){ObjectInputStream ois = new ObjectInputStream(in);Object o = ois.readObject();System.out.println(name + "check result：" + o.toString());}else {TimeUnit.MILLISECONDS.sleep(10);}}catch (Exception e){e.printStackTrace();}}}}
}

2.2 Server.java

package com.demo.longconn.server;import com.demo.longconn.KeepAlive;import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;/*** User: lanxinghua* Date: 2019/11/4 18:49* Desc: 服务端,服务端检测receiveTimeDelay内没接收到任何数据，自动和客户端断开连接。*/
public class Server {private int port;private ServerSocket serverSocket;private Socket socket;// 连接状态private volatile boolean isConn = false;// 检测心跳时间 3sprivate long receiveTimeDelay = 3000;// 最后一次接收时间long lastReceiveTime;// 有效客户端private static Map<String/*clientName*/, ClientState> map = new HashMap<String, ClientState>(10);private Object lock = new Object();public Server(int port) {this.port = port;}public void start(){if (isConn) return;try {System.out.println("服务端启动......");isConn = true;// 检测客户端心跳new Thread(new ConnectWatchDog()).start();}catch (Exception e){e.printStackTrace();stop();}}public void stop(){if (isConn) isConn = false;}/*** 连接监控*/class ConnectWatchDog implements Runnable{public void run() {try {serverSocket = new ServerSocket(port);while (isConn){socket = serverSocket.accept();lastReceiveTime = System.currentTimeMillis();new Thread(new SocketAction(socket)).start();}}catch (Exception e){e.printStackTrace();stop();}}}/*** 监控客户端宕机的机器*/class CleanScan implements Runnable{public void run() {while (true) {synchronized (lock) {if (map.isEmpty()) {return;}for (Map.Entry<String, ClientState> client : map.entrySet()) {ClientState clientState = client.getValue();if (clientState == null) {map.remove(client.getKey());return;}if (clientState.getIsValid() == 0) {continue;}if (System.currentTimeMillis() - clientState.getLastReTime() > receiveTimeDelay) {clientState.setIsValid(0);System.out.println("有效客户端" + getAliveClientCount() +  " 客户端宕机" + clientState.toString());}}}try {TimeUnit.SECONDS.sleep(3);}catch (Exception e){e.printStackTrace();}}}/*** 获取有效客户端数量* @return*/private long getAliveClientCount(){return map.values().stream().filter(e -> e.getIsValid() == 1).count();}}class SocketAction implements Runnable{boolean isRun = true;Socket s;ClientState clientState;public SocketAction(Socket s) {this.s = s;clientState = new ClientState();}public void run() {try {while (isConn && isRun){new Thread(new CleanScan()).start();if (System.currentTimeMillis() - lastReceiveTime > receiveTimeDelay){close();}else {InputStream in = s.getInputStream();if (in.available() > 0){ObjectInputStream ois = new ObjectInputStream(in);Object o = ois.readObject();lastReceiveTime = System.currentTimeMillis();if (o instanceof KeepAlive){KeepAlive alive = (KeepAlive) o;System.out.println("客户端数量：" + getAliveClientCount() + " "+alive.getClientName() + "  心跳检查ok：" + o.toString());clientState.setClientName(alive.getClientName());clientState.setIsValid(1);clientState.setLastReTime(lastReceiveTime);map.put(alive.getClientName(), clientState);}}else {TimeUnit.MILLISECONDS.sleep(10);}}}}catch (Exception e){e.printStackTrace();close();}}private long getAliveClientCount(){return map.values().stream().filter(e -> e.getIsValid() == 1).count();}private void close(){if (isRun) isRun = false;if (socket!=null){try {socket.close();}catch (Exception e){e.printStackTrace();}}}}
}

2.3 ClientState.java

package com.demo.longconn.server;/*** @Author: cxx* @Date: 2019/11/4 22:34* 客户端服务器状态*/
public class ClientState {private int isValid = 0;private String clientName;private long lastReTime;public int getIsValid() {return isValid;}public void setIsValid(int isValid) {this.isValid = isValid;}public String getClientName() {return clientName;}public void setClientName(String clientName) {this.clientName = clientName;}public long getLastReTime() {return lastReTime;}public void setLastReTime(long lastReTime) {this.lastReTime = lastReTime;}@Overridepublic String toString() {return "ClientState{" +"isValid=" + isValid +", clientName='" + clientName + '\'' +", lastReTime=" + lastReTime +'}';}
}

2.4 KeepAlive.java

package com.demo.longconn;import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Date;/*** User: lanxinghua* Date: 2019/11/4 18:45* Desc: 维持连接的消息对象，心跳对象*/
public class KeepAlive implements Serializable {private String clientName;public KeepAlive(String clientName) {this.clientName = clientName;}@Overridepublic String toString() {return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()) + "\t维持连接包";}public String getClientName() {return clientName;}public void setClientName(String clientName) {this.clientName = clientName;}
}

2.5 ServerTest.java

package com.demo.longconn;import com.demo.longconn.server.Server;/*** User: lanxinghua* Date: 2019/11/4 19:32* Desc:*/
public class ServerTest {public static void main(String[] args) {new Server(9999).start();}
}

2.6 ClientTest.java

package com.demo.longconn;import com.demo.longconn.client.Client;/*** User: lanxinghua* Date: 2019/11/4 19:32* Desc:*/
public class ClientTest {public static void main(String[] args) {for (int i = 1; i <=3; i++) {new Client("localhost", 9999, "客户端" + i).start();}}
}

三、测试

3.1 开启服务端

服务端启动…

3.2 开启客户端

客户端1已启动…
客户端2已启动…
客户端3已启动…

测试计划：假设客户端宕机

// 假设客户端宕机
if ("客户端3".equals(name)){System.out.println(name + "出现故障，将在1s后宕机");TimeUnit.SECONDS.sleep(1);stop();
}if ("客户端2".equals(name)){System.out.println(name + "出现故障，将在2s后宕机");TimeUnit.SECONDS.sleep(2);stop();
}

第九篇 - 手写心跳检测相关推荐

书接上文，基于藏文手写数字数据开发构建yolov5n轻量级藏文手写数字检测识别系统
在上一篇文章中: <python基于轻量级CNN模型开发构建手写藏文数字识别系统> 开发实现了轻量级的藏文手写数字识别系统,这里主要是想基于前文的数据,整合目标检测模型来进一步挖掘藏文手写 ...
计算机视觉 || 手写字体检测
手写字体的检测 (1) Adaboost进行手写字体的检测导入mnist数据集 import tensorflow.examples.tutorials.mnist.input_data as in ...
第六篇 - 手写基于接口实现动态代理
Github源码下载地址:https://github.com/chenxingxing6/sourcecode/tree/master/code-proxy 一.前言我们知道常见的动态代理有两种实 ...
第五篇 - 手写Tomcat(基于Netty)热部署
Github源码下载地址:https://github.com/chenxingxing6/sourcecode/tree/master/code-netty-tomcat 一.前言 Tomcat是一 ...
iOS开发UI篇—手写控件，frame，center和bounds属性
一.手写控件 1.手写控件的步骤 (1)使用相应的控件类创建控件对象 (2)设置该控件的各种属性 (3)添加控件到视图中 (4)如果是button等控件,还需考虑控件的单击事件等 (5)注意:View ...
第四篇 - 手写RPC框架
Github源码下载地址:https://github.com/chenxingxing6/myrpc 一.前言 RPC(Remote Procedure Call)-远程过程调用,它是一种通过网络从 ...
第一篇 - 手写SpringMvc框架
Github源码下载地址:https://github.com/chenxingxing6/springmvc CSDN源码下载地址:https://download.csdn.net/downloa ...
第02篇:手写JavaRPC框架之设计思路
作者: 西魏陶渊明博客: https://blog.springlearn.cn/ 天下代码一大抄, 抄来抄去有提高, 看你会抄不会抄! 一.前言隔壁老李又在喷我了: "完犊子了,小编这 ...
2020年汤家凤直播讲解1800题基础篇手写笔记-不定积分和定积分部分
写在前面:下面是汤家凤老师直播讲解1800题中不定积分.定积分部分的笔记,笔者已抄题并进行思路和答案记录. 文章目录 1 不定积分的计算 2 定积分三角函数.Γ函数.周期函数的结论 1800 基础篇 ...