快捷導(dǎo)航

JAVA實現(xiàn)心跳檢測(長連接)

更新時間：2023年03月01日 10:41:15 作者：菜鳥---程序員

本文主要介紹了JAVA實現(xiàn)心跳檢測(長連接)，文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧

1、心跳機(jī)制簡介

在分布式系統(tǒng)中，分布在不同主機(jī)上的節(jié)點需要檢測其他節(jié)點的狀態(tài)，如服務(wù)器節(jié)點需要檢測從節(jié)點是否失效。為了檢測對方節(jié)點的有效性，每隔固定時間就發(fā)送一個固定信息給對方，對方回復(fù)一個固定信息，如果長時間沒有收到對方的回復(fù)，則斷開與對方的連接。

發(fā)包方既可以是服務(wù)端，也可以是客戶端，這要看具體實現(xiàn)。因為是每隔固定時間發(fā)送一次，類似心跳，所以發(fā)送的固定信息稱為心跳包。心跳包一般為比較小的包，可根據(jù)具體實現(xiàn)。心跳包主要應(yīng)用于長連接的保持與短線鏈接。

一般而言，應(yīng)該客戶端主動向服務(wù)器發(fā)送心跳包，因為服務(wù)器向客戶端發(fā)送心跳包會影響服務(wù)器的性能。

實現(xiàn)原理：長連接的維持，是要客戶端程序，定時向服務(wù)端程序，發(fā)送一個維持連接包的。如果，長時間未發(fā)送維持連接包，服務(wù)端程序?qū)嚅_連接。

2、心跳機(jī)制實現(xiàn)方式

心跳機(jī)制有兩種實現(xiàn)方式，一種基于TCP自帶的心跳包，TCP的SO_KEEPALIVE選項可以，系統(tǒng)默認(rèn)的默認(rèn)跳幀頻率為2小時，超過2小時后，本地的TCP 實現(xiàn)會發(fā)送一個數(shù)據(jù)包給遠(yuǎn)程的 Socket. 如果遠(yuǎn)程Socket 沒有發(fā)回響應(yīng), TCP實現(xiàn)就會持續(xù)嘗試 11 分鐘, 直到接收到響應(yīng)為止。否則就會自動斷開Socket連接。但TCP自帶的心跳包無法檢測比較敏感地知道對方的狀態(tài)，默認(rèn)2小時的空閑時間，對于大多數(shù)的應(yīng)用而言太長了。可以手工開啟KeepAlive功能并設(shè)置合理的KeepAlive參數(shù)。

另一種在應(yīng)用層自己進(jìn)行實現(xiàn)，基本步驟如下：

1、Client使用定時器，不斷發(fā)送心跳；

2、Server收到心跳后，回復(fù)一個包；

3、Server為每個Client啟動超時定時器，如果在指定時間內(nèi)沒有收到Client的心跳包，則Client失效。

3、客戶端

Client通過持有Socket的對象，可以隨時（使用sendObject方法）發(fā)送Massage Object(消息)給服務(wù)端。

如果keepAliveDelay毫秒（程序中是2秒）內(nèi)未發(fā)送任何數(shù)據(jù)，則自動發(fā)送一個KeepAlive Object(心跳)給服務(wù)端，用于維持連接。

由于，我們向服務(wù)端，可以發(fā)送很多不同的消息對象，服務(wù)端也可以返回不同的對象。所以，對于返回對象的處理，要編寫具體的ObjectAction實現(xiàn)類進(jìn)行處理。通過Client.addActionMap方法進(jìn)行添加。這樣，程序會回調(diào)處理

4 、服務(wù)端

由于客戶端會定時（keepAliveDelay毫秒）發(fā)送維持連接的信息過來，所以，服務(wù)端要有一個檢測機(jī)制。

即當(dāng)服務(wù)端receiveTimeDelay毫秒（程序中是3秒）內(nèi)未接收任何數(shù)據(jù)，則自動斷開與客戶端的連接。

ActionMapping的原理與客戶端相似（相同）。

通過添加相應(yīng)的ObjectAction實現(xiàn)類，可以實現(xiàn)不同對象的響應(yīng)、應(yīng)答過程。

5、代碼實現(xiàn)

5.1 KeepAlive.java

package com.zyz.mynative.demo02;

/**
 * @author zyz
 * @version 1.0
 * @data 2023/2/15 11:05
 * @Description:
 */
import java.io.Serializable;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 *
 * 維持連接的消息對象（心跳對象）
 * @author zyz
 * @version 1.0
 * @data 2023/2/15 11:05
 * @Description:
 */
public class KeepAlive implements Serializable{

    private static final long serialVersionUID = -2813120366138988480L;

    /* 覆蓋該方法，僅用于測試使用。
     * @see java.lang.Object#toString()
     */

    @Override
    public String toString() {
        return new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())+"\t維持連接包";
    }

}

5.2 MyClient.java

package com.zyz.mynative.demo02;

/**
 * @author zyz
 * @version 1.0
 * @data 2023/2/15 11:06
 * @Description:
 */


import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.Socket;
import java.net.UnknownHostException;
import java.util.concurrent.ConcurrentHashMap;

/**
 * C/S架構(gòu)的客戶端對象，持有該對象，可以隨時向服務(wù)端發(fā)送消息。
 *
 * @author zyz
 * @version 1.0
 * @data 2023/2/15 11:06
 * @Description:
 */
public class MyClient {

    /**
     * 處理服務(wù)端發(fā)回的對象，可實現(xiàn)該接口。
     */
    public static interface ObjectAction {
        void doAction(Object obj, MyClient client);
    }

    public static final class DefaultObjectAction implements ObjectAction {
        @Override
        public void doAction(Object obj, MyClient client) {
            System.out.println("處理：\t" + obj.toString());
        }
    }


    private String serverIp;
    private int port;
    private Socket socket;
    //連接狀態(tài)
    private boolean running = false;

    /**
     * 最后一次發(fā)送數(shù)據(jù)的時間
     */
    private long lastSendTime;

    /**
     * 用于保存接收消息對象類型及該類型消息處理的對象
     */
    private ConcurrentHashMap<Class, ObjectAction> actionMapping = new ConcurrentHashMap<Class, ObjectAction>();

    public MyClient(String serverIp, int port) {
        this.serverIp = serverIp;
        this.port = port;
    }

    public void start() throws UnknownHostException, IOException {
        if (running) {
            return;
        }
        socket = new Socket(serverIp, port);
        System.out.println("本地端口：" + socket.getLocalPort());
        lastSendTime = System.currentTimeMillis();
        running = true;

        //保持長連接的線程，每隔2秒項服務(wù)器發(fā)一個一個保持連接的心跳消息
        new Thread(new KeepAliveWatchDog()).start();
        //接受消息的線程，處理消息
        new Thread(new ReceiveWatchDog()).start();
    }

    public void stop() {
        if (running) {
            running = false;
        }
        ;
    }

    /**
     * 添加接收對象的處理對象。
     *
     * @param cls    待處理的對象，其所屬的類。
     * @param action 處理過程對象。
     */
    public void addActionMap(Class<Object> cls, ObjectAction action) {
        actionMapping.put(cls, action);
    }

    public void sendObject(Object obj) throws IOException {
        ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
        oos.writeObject(obj);
        System.out.println("發(fā)送：\t" + obj);
        oos.flush();
    }



    class KeepAliveWatchDog implements Runnable {
        long checkDelay = 10;
        //兩秒鐘檢測一次
        long keepAliveDelay = 2000;

        @Override
        public void run() {
            while (running) {
                if (System.currentTimeMillis() - lastSendTime > keepAliveDelay) {
                    try {
                        MyClient.this.sendObject(new KeepAlive());
                    } catch (IOException e) {
                        e.printStackTrace();
                        MyClient.this.stop();
                    }
                    lastSendTime = System.currentTimeMillis();
                } else {
                    try {
                        Thread.sleep(checkDelay);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        MyClient.this.stop();
                    }
                }
            }
        }
    }

    class ReceiveWatchDog implements Runnable {
        @Override
        public void run() {
            while (running) {
                try {
                    InputStream in = socket.getInputStream();
                    if (in.available() > 0) {
                        ObjectInputStream ois = new ObjectInputStream(in);
                        Object obj = ois.readObject();
                        System.out.println("接收：\t" + obj);
                        ObjectAction oa = actionMapping.get(obj.getClass());
                        oa = oa == null ? new DefaultObjectAction() : oa;
                        oa.doAction(obj, MyClient.this);
                    } else {
                        Thread.sleep(10);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                    MyClient.this.stop();
                }
            }
        }
    }


    public static void main(String[] args) throws UnknownHostException, IOException {
        String serverIp = "127.0.0.1";
        int port = 65432;
        MyClient client = new MyClient(serverIp, port);
        client.start();
    }
}

5.3 MyServer

package com.zyz.mynative.demo02;

import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.ConcurrentHashMap;

/**
 * C/S架構(gòu)的服務(wù)端對象。
 *
 * @author zyz
 * @version 1.0
 * @data 2023/2/15 11:08
 * @Description:
 */
public class MyServer {


    /**
     * 要處理客戶端發(fā)來的對象，并返回一個對象，可實現(xiàn)該接口。
     */
    public interface ObjectAction {
        Object doAction(Object rev, MyServer server);
    }

    public static final class DefaultObjectAction implements ObjectAction {
        @Override
        public Object doAction(Object rev, MyServer server) {
            System.out.println("處理并返回：" + rev);
            return rev;
        }
    }


    private int port;
    private volatile boolean running = false;
    private long receiveTimeDelay = 3000;
    /**
     * ConcurrentHashMap是線程安全的，ConcurrentHashMap并非鎖住整個方法，
     * 而是通過原子操作和局部加鎖的方法保證了多線程的線程安全，且盡可能減少了性能損耗。
     */
    private ConcurrentHashMap<Class, ObjectAction> actionMapping = new ConcurrentHashMap<Class, ObjectAction>();
    private Thread connWatchDog;

    public MyServer(int port) {
        this.port = port;
    }

    /**
     * 通過繼承Runnable ,開啟線程
     */
    public void start() {
        if (running) {
            return;
        }
        ;
        running = true;
        connWatchDog = new Thread(new ConnWatchDog());
        connWatchDog.start();
    }

    @SuppressWarnings("deprecation")
    public void stop() {
        if (running) {
            running = false;
        }
        ;
        if (connWatchDog != null) {
            connWatchDog.stop();
        }
        ;
    }

    public void addActionMap(Class<Object> cls, ObjectAction action) {
        actionMapping.put(cls, action);
    }

    /**
     * 繼承Runnable 重寫run方法、實現(xiàn)線程
     */
    class ConnWatchDog implements Runnable {
        @Override
        public void run() {
            try {
                ServerSocket ss = new ServerSocket(port, 5);
                while (running) {
                    Socket s = ss.accept();
                    new Thread(new SocketAction(s)).start();
                }
            } catch (IOException e) {
                e.printStackTrace();
                MyServer.this.stop();
            }

        }
    }

    /**
     * 繼承Runnable 重寫run方法、實現(xiàn)線程，通過匿名內(nèi)部類
     */
    class SocketAction implements Runnable {
        Socket s;
        boolean run = true;
        long lastReceiveTime = System.currentTimeMillis();

        public SocketAction(Socket s) {
            this.s = s;
        }

        @Override
        public void run() {
            while (running && run) {
                if (System.currentTimeMillis() - lastReceiveTime > receiveTimeDelay) {
                    overThis();
                } else {
                    try {
                        InputStream in = s.getInputStream();
                        if (in.available() > 0) {
                            ObjectInputStream ois = new ObjectInputStream(in);
                            Object obj = ois.readObject();
                            lastReceiveTime = System.currentTimeMillis();
                            System.out.println("接收：\t" + obj);
                            ObjectAction oa = actionMapping.get(obj.getClass());
                            oa = oa == null ? new DefaultObjectAction() : oa;
                            Object out = oa.doAction(obj, MyServer.this);
                            if (out != null) {
                                ObjectOutputStream oos = new ObjectOutputStream(s.getOutputStream());
                                oos.writeObject(out);
                                oos.flush();
                            }
                        } else {
                            Thread.sleep(10);
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        overThis();
                    }
                }
            }
        }

        private void overThis() {
            if (run) {
                run = false;
            }
            ;
            if (s != null) {
                try {
                    s.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            System.out.println("關(guān)閉：" + s.getRemoteSocketAddress());
        }

    }

    public static void main(String[] args) {
        int port = 65432;
        MyServer server = new MyServer(port);
        server.start();
    }

}