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

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

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

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

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

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

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

Client使用定時器，不斷發(fā)送心跳；Server收到心跳后，回復(fù)一個包；Server為每個Client啟動超時定時器，如果在指定時間內(nèi)沒有收到Client的心跳包，則Client失效。

三、Java實(shí)現(xiàn)心跳機(jī)制

這里基于Java實(shí)現(xiàn)的簡單RPC框架實(shí)現(xiàn)心跳機(jī)制。Java實(shí)現(xiàn)代碼如下所示：

心跳客戶端類：

public class HeartbeatClient implements Runnable { private String serverIP = '127.0.0.1'; private int serverPort = 8089; private String nodeID = UUID.randomUUID().toString(); private boolean isRunning = true; // 最近的心跳時間 private long lastHeartbeat; // 心跳間隔時間 private long heartBeatInterval = 10 * 1000; public void run() { try { while (isRunning) { HeartbeatHandler handler = RPClient.getRemoteProxyObj(HeartbeatHandler.class, new InetSocketAddress(serverIP, serverPort)); long startTime = System.currentTimeMillis(); // 是否達(dá)到發(fā)送心跳的周期時間 if (startTime - lastHeartbeat > heartBeatInterval) { System.out.println('send a heart beat'); lastHeartbeat = startTime; HeartbeatEntity entity = new HeartbeatEntity(); entity.setTime(startTime); entity.setNodeID(nodeID); // 向服務(wù)器發(fā)送心跳，并返回需要執(zhí)行的命令 Cmder cmds = handler.sendHeartBeat(entity); if (!processCommand(cmds)) continue; } } } catch (Exception e) { e.printStackTrace(); } } private boolean processCommand(Cmder cmds) { // ... return true; } }

心跳包實(shí)體類：

public class HeartbeatEntity implements Serializable { private long time; private String nodeID; private String error; private Map<String, Object> info = new HashMap<String, Object>(); public String getNodeID() { return nodeID; } public void setNodeID(String nodeID) { this.nodeID = nodeID; } public String getError() { return error; } public void setError(String error) { this.error = error; } public Map<String, Object> getInfo() { return info; } public void setInfo(Map<String, Object> info) { this.info = info; } public long getTime() { return time; } public void setTime(long time) { this.time = time; }}

服務(wù)器接受心跳包返回的命令對象類：

public class Cmder implements Serializable { private String nodeID; private String error; private Map<String, Object> info = new HashMap<String, Object>(); public String getNodeID() { return nodeID; } public void setNodeID(String nodeID) { this.nodeID = nodeID; } public String getError() { return error; } public void setError(String error) { this.error = error; } public Map<String, Object> getInfo() { return info; } public void setInfo(Map<String, Object> info) { this.info = info; }}

RPC服務(wù)注冊中心：

public class ServiceCenter { private ExecutorService executor = Executors.newFixedThreadPool(20); private final ConcurrentHashMap<String, Class> serviceRegistry = new ConcurrentHashMap<String, Class>(); private AtomicBoolean isRunning = new AtomicBoolean(true); // 服務(wù)器監(jiān)聽端口 private int port = 8089; // 心跳監(jiān)聽器 HeartbeatLinstener linstener; // 單例模式 private static class SingleHolder { private static final ServiceCenter INSTANCE = new ServiceCenter(); } private ServiceCenter() { } public static ServiceCenter getInstance() { return SingleHolder.INSTANCE; } public void register(Class serviceInterface, Class impl) { System.out.println('regeist service ' + serviceInterface.getName()); serviceRegistry.put(serviceInterface.getName(), impl); } public void start() throws IOException { ServerSocket server = new ServerSocket(); server.bind(new InetSocketAddress(port)); System.out.println('start server'); linstener = HeartbeatLinstener.getInstance(); System.out.println('start listen heart beat'); try { while (true) { // 1.監(jiān)聽客戶端的TCP連接，接到TCP連接后將其封裝成task，由線程池執(zhí)行 executor.execute(new ServiceTask(server.accept())); } } finally { server.close(); } } public void stop() { isRunning.set(false); executor.shutdown(); } public boolean isRunning() { return isRunning.get(); } public int getPort() { return port; } public void settPort(int port) { this.port = port; } public ConcurrentHashMap<String, Class> getServiceRegistry() { return serviceRegistry; } private class ServiceTask implements Runnable { Socket clent = null; public ServiceTask(Socket client) { this.clent = client; } public void run() { ObjectInputStream input = null; ObjectOutputStream output = null; try { // 2.將客戶端發(fā)送的碼流反序列化成對象，反射調(diào)用服務(wù)實(shí)現(xiàn)者，獲取執(zhí)行結(jié)果 input = new ObjectInputStream(clent.getInputStream()); String serviceName = input.readUTF(); String methodName = input.readUTF(); Class<?>[] parameterTypes = (Class<?>[]) input.readObject(); Object[] arguments = (Object[]) input.readObject(); Class serviceClass = serviceRegistry.get(serviceName); if (serviceClass == null) { throw new ClassNotFoundException(serviceName + ' not found'); } Method method = serviceClass.getMethod(methodName, parameterTypes); Object result = method.invoke(serviceClass.newInstance(), arguments); // 3.將執(zhí)行結(jié)果反序列化，通過socket發(fā)送給客戶端 output = new ObjectOutputStream(clent.getOutputStream()); output.writeObject(result); } catch (Exception e) { e.printStackTrace(); } finally { if (output != null) { try { output.close(); } catch (IOException e) { e.printStackTrace(); } } if (input != null) { try { input.close(); } catch (IOException e) { e.printStackTrace(); } } if (clent != null) { try { clent.close(); } catch (IOException e) { e.printStackTrace(); } } } } }}

心跳監(jiān)聽類：

package com.cang.heartbeat; import java.io.IOException;import java.io.ObjectInputStream;import java.io.ObjectOutputStream;import java.lang.reflect.Method;import java.net.InetSocketAddress;import java.net.ServerSocket;import java.net.Socket;import java.util.Iterator;import java.util.Map;import java.util.concurrent.ConcurrentHashMap;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;import java.util.concurrent.atomic.AtomicBoolean; /** * 心跳監(jiān)聽保存信息 * * @author cang * @create_time 2016-09-28 11:40 */public class HeartbeatLinstener { private ExecutorService executor = Executors.newFixedThreadPool(20); private final ConcurrentHashMap<String, Object> nodes = new ConcurrentHashMap<String, Object>(); private final ConcurrentHashMap<String, Long> nodeStatus = new ConcurrentHashMap<String, Long>(); private long timeout = 10 * 1000; // 服務(wù)器監(jiān)聽端口 private int port = 8089; // 單例模式 private static class SingleHolder { private static final HeartbeatLinstener INSTANCE = new HeartbeatLinstener(); } private HeartbeatLinstener() { } public static HeartbeatLinstener getInstance() { return SingleHolder.INSTANCE; } public ConcurrentHashMap<String, Object> getNodes() { return nodes; } public void registerNode(String nodeId, Object nodeInfo) { nodes.put(nodeId, nodeInfo); nodeStatus.put(nodeId, System.currentTimeMillis()); } public void removeNode(String nodeID) { if (nodes.containsKey(nodeID)) { nodes.remove(nodeID); } } // 檢測節(jié)點(diǎn)是否有效 public boolean checkNodeValid(String key) { if (!nodes.containsKey(key) || !nodeStatus.containsKey(key)) return false; if ((System.currentTimeMillis() - nodeStatus.get(key)) > timeout) return false; return true; } // 刪除所有失效節(jié)點(diǎn) public void removeInValidNode() { Iterator<Map.Entry<String, Long>> it = nodeStatus.entrySet().iterator(); while (it.hasNext()) { Map.Entry<String, Long> e = it.next(); if ((System.currentTimeMillis() - nodeStatus.get(e.getKey())) > timeout) { nodes.remove(e.getKey()); } } } }

心跳處理類接口：

public interface HeartbeatHandler { public Cmder sendHeartBeat(HeartbeatEntity info);}

心跳處理實(shí)現(xiàn)類：

public class HeartbeatHandlerImpl implements HeartbeatHandler { public Cmder sendHeartBeat(HeartbeatEntity info) { HeartbeatLinstener linstener = HeartbeatLinstener.getInstance(); // 添加節(jié)點(diǎn) if (!linstener.checkNodeValid(info.getNodeID())) { linstener.registerNode(info.getNodeID(), info); } // 其他操作 Cmder cmder = new Cmder(); cmder.setNodeID(info.getNodeID()); // ... System.out.println('current all the nodes: '); Map<String, Object> nodes = linstener.getNodes(); for (Map.Entry e : nodes.entrySet()) { System.out.println(e.getKey() + ' : ' + e.getValue()); } System.out.println('hadle a heartbeat'); return cmder; }}

測試類：

public class HeartbeatTest { public static void main(String[] args) { new Thread(new Runnable() { public void run() { try { ServiceCenter serviceServer = ServiceCenter.getInstance(); serviceServer.register(HeartbeatHandler.class, HeartbeatHandlerImpl.class); serviceServer.start(); } catch (IOException e) { e.printStackTrace(); } } }).start(); Thread client1 = new Thread(new HeartbeatClient()); client1.start(); Thread client2 = new Thread(new HeartbeatClient()); client2.start(); }}

四、總結(jié)

上面的代碼還有很多不足的地方,希望有空能進(jìn)行改善：

其他小問題就暫時不管了，希望以后能重寫上面的代碼。

以上就是Java實(shí)現(xiàn)心跳機(jī)制的方法的詳細(xì)內(nèi)容，更多關(guān)于Java實(shí)現(xiàn)心跳機(jī)制的資料請關(guān)注好吧啦網(wǎng)其它相關(guān)文章！