2. 程式人生 > >NETTY 心跳機制

NETTY 心跳機制

阿新 • • 發佈：2018-12-13
end value trap parseint acceptor blog channel 出發 每次


最近工作比較忙，但閑暇之余還是看了阿裏的馮家春（fengjiachun）的github上的開源代碼Jupiter，寫的RPC框架讓我感嘆人外有人，廢話不多說，下面的代碼全部截取自Jupiter，寫了一個比較完整的例子，供大家一起學習分享，再次對@Luca抱拳，Jupiter的Github地址：

https://github.com/fengjiachun/Jupiter

今天研究的是，心跳和重連，雖然這次是大神寫的代碼，但是萬變不離其宗，我們先回顧一下Netty應用心跳和重連的整個過程：

1）客戶端連接服務端

2）在客戶端的的ChannelPipeline中加入一個比較特殊的IdleStateHandler，設置一下客戶端的寫空閑時間，例如5s

3）當客戶端的所有ChannelHandler中4s內沒有write事件，則會觸發userEventTriggered方法（上文介紹過）

4）我們在客戶端的userEventTriggered中對應的觸發事件下發送一個心跳包給服務端，檢測服務端是否還存活，防止服務端已經宕機，客戶端還不知道

5）同樣，服務端要對心跳包做出響應，其實給客戶端最好的回復就是“不回復”，這樣可以服務端的壓力，假如有10w個空閑Idle的連接，那麽服務端光發送心跳回復，則也是費事的事情，那麽怎麽才能告訴客戶端它還活著呢，其實很簡單，因為5s服務端都會收到來自客戶端的心跳信息，那麽如果10秒內收不到，服務端可以認為客戶端掛了，可以close鏈路

6）加入服務端因為什麽因素導致宕機的話，就會關閉所有的鏈路鏈接，所以作為客戶端要做的事情就是短線重連

以上描述的就是整個心跳和重連的整個過程，雖然很簡單，上一篇blog也寫了一個Demo,簡單地做了一下上述功能

要寫工業級的Netty心跳重連的代碼，需要解決一下幾個問題：

1）ChannelPipeline中的ChannelHandlers的維護，首次連接和重連都需要對ChannelHandlers進行管理

2）重連對象的管理，也就是bootstrap對象的管理

3）重連機制編寫

完整的代碼：https://github.com/BazingaLyn/netty-study/tree/master/src/main/java/com/lyncc/netty/idle

下面我們就看大神是如何解決這些問題的，首先先定義一個接口ChannelHandlerHolder，用來保管ChannelPipeline中的Handlers的

package com.lyncc.netty.idle;

import io.netty.channel.ChannelHandler;

/**
*
* 客戶端的ChannelHandler集合，由子類實現，這樣做的好處：
* 繼承這個接口的所有子類可以很方便地獲取ChannelPipeline中的Handlers
* 獲取到handlers之後方便ChannelPipeline中的handler的初始化和在重連的時候也能很方便
* 地獲取所有的handlers
*/
public interface ChannelHandlerHolder {

ChannelHandler[] handlers();
}
我們再來編寫我們熟悉的服務端的ServerBootstrap的編寫：
HeartBeatServer.java

package com.lyncc.netty.idle;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

public class HeartBeatServer {

private final AcceptorIdleStateTrigger idleStateTrigger = new AcceptorIdleStateTrigger();

private int port;

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

public void start() {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap sbs = new ServerBootstrap().group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO))
.localAddress(new InetSocketAddress(port)).childHandler(new ChannelInitializer<SocketChannel>() {
protected void initChannel(SocketChannel ch) throws Exception {
ch.pipeline().addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));
ch.pipeline().addLast(idleStateTrigger);
ch.pipeline().addLast("decoder", new StringDecoder());
ch.pipeline().addLast("encoder", new StringEncoder());
ch.pipeline().addLast(new HeartBeatServerHandler());
};

}).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);
// 綁定端口，開始接收進來的連接
ChannelFuture future = sbs.bind(port).sync();

System.out.println("Server start listen at " + port);
future.channel().closeFuture().sync();
} catch (Exception e) {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}

public static void main(String[] args) throws Exception {
int port;
if (args.length > 0) {
port = Integer.parseInt(args[0]);
} else {
port = 8080;
}
new HeartBeatServer(port).start();
}

}
單獨寫一個AcceptorIdleStateTrigger，其實也是繼承ChannelInboundHandlerAdapter，重寫userEventTriggered方法，因為客戶端是write，那麽服務端自然是read，設置的狀態就是IdleState.READER_IDLE，源碼如下：
package com.lyncc.netty.idle;

import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;


@ChannelHandler.Sharable
public class AcceptorIdleStateTrigger extends ChannelInboundHandlerAdapter {

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleState state = ((IdleStateEvent) evt).state();
if (state == IdleState.READER_IDLE) {
throw new Exception("idle exception");
}
} else {
super.userEventTriggered(ctx, evt);
}
}
}
HeartBeatServerHandler就是一個很簡單的自定義的Handler，不是重點：
package com.lyncc.netty.idle;

import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class HeartBeatServerHandler extends ChannelInboundHandlerAdapter {


@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
System.out.println("server channelRead..");
System.out.println(ctx.channel().remoteAddress() + "->Server :" + msg.toString());
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}

}
接下來就是重點，我們需要寫一個類，這個類可以去觀察鏈路是否斷了，如果斷了，進行循環的斷線重連操作，ConnectionWatchdog，顧名思義，鏈路檢測狗，我們先看完整代碼：

package com.lyncc.netty.idle;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.util.Timeout;
import io.netty.util.Timer;
import io.netty.util.TimerTask;

import java.util.concurrent.TimeUnit;

/**
*
* 重連檢測狗，當發現當前的鏈路不穩定關閉之後，進行12次重連
*/
@Sharable
public abstract class ConnectionWatchdog extends ChannelInboundHandlerAdapter implements TimerTask ,ChannelHandlerHolder{



private final Bootstrap bootstrap;
private final Timer timer;
private final int port;

private final String host;

private volatile boolean reconnect = true;
private int attempts;


public ConnectionWatchdog(Bootstrap bootstrap, Timer timer, int port,String host, boolean reconnect) {
this.bootstrap = bootstrap;
this.timer = timer;
this.port = port;
this.host = host;
this.reconnect = reconnect;
}

/**
* channel鏈路每次active的時候，將其連接的次數重新? 0
*/
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {

System.out.println("當前鏈路已經激活了，重連嘗試次數重新置為0");

attempts = 0;
ctx.fireChannelActive();
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("鏈接關閉");
if(reconnect){
System.out.println("鏈接關閉，將進行重連");
if (attempts < 12) {
attempts++;
//重連的間隔時間會越來越長
int timeout = 2 << attempts;
timer.newTimeout(this, timeout, TimeUnit.MILLISECONDS);
}
}
ctx.fireChannelInactive();
}


public void run(Timeout timeout) throws Exception {

ChannelFuture future;
//bootstrap已經初始化好了，只需要將handler填入就可以了
synchronized (bootstrap) {
bootstrap.handler(new ChannelInitializer<Channel>() {

@Override
protected void initChannel(Channel ch) throws Exception {

ch.pipeline().addLast(handlers());
}
});
future = bootstrap.connect(host,port);
}
//future對象
future.addListener(new ChannelFutureListener() {

public void operationComplete(ChannelFuture f) throws Exception {
boolean succeed = f.isSuccess();

//如果重連失敗，則調用ChannelInactive方法，再次出發重連事件，一直嘗試12次，如果失敗則不再重連
if (!succeed) {
System.out.println("重連失敗");
f.channel().pipeline().fireChannelInactive();
}else{
System.out.println("重連成功");
}
}
});

}

}


稍微分析一下：
1）繼承了ChannelInboundHandlerAdapter，說明它也是Handler，也對，作為一個檢測對象，肯定會放在鏈路中，否則怎麽檢測

2）實現了2個接口，TimeTask，ChannelHandlerHolder

①TimeTask，我們就要寫run方法，這應該是一個定時任務，這個定時任務做的事情應該是重連的工作

②ChannelHandlerHolder的接口，這個接口我們剛才說過是維護的所有的Handlers，因為在重連的時候需要獲取Handlers

3）bootstrap對象，重連的時候依舊需要這個對象

4）當鏈路斷開的時候會觸發channelInactive這個方法，也就說觸發重連的導火索是從這邊開始的

好了，我們這邊再寫次核心的HeartBeatsClient的代碼：

package com.lyncc.netty.idle;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.HashedWheelTimer;

import java.util.concurrent.TimeUnit;

public class HeartBeatsClient {

protected final HashedWheelTimer timer = new HashedWheelTimer();

private Bootstrap boot;

private final ConnectorIdleStateTrigger idleStateTrigger = new ConnectorIdleStateTrigger();

public void connect(int port, String host) throws Exception {

EventLoopGroup group = new NioEventLoopGroup();

boot = new Bootstrap();
boot.group(group).channel(NioSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO));

final ConnectionWatchdog watchdog = new ConnectionWatchdog(boot, timer, port,host, true) {

public ChannelHandler[] handlers() {
return new ChannelHandler[] {
this,
new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS),
idleStateTrigger,
new StringDecoder(),
new StringEncoder(),
new HeartBeatClientHandler()
};
}
};

ChannelFuture future;
//進行連接
try {
synchronized (boot) {
boot.handler(new ChannelInitializer<Channel>() {

//初始化channel
@Override
protected void initChannel(Channel ch) throws Exception {
ch.pipeline().addLast(watchdog.handlers());
}
});

future = boot.connect(host,port);
}

// 以下代碼在synchronized同步塊外面是安全的
future.sync();
} catch (Throwable t) {
throw new Exception("connects to fails", t);
}
}

/**
* @param args
* @throws Exception
*/
public static void main(String[] args) throws Exception {
int port = 8080;
if (args != null && args.length > 0) {
try {
port = Integer.valueOf(args[0]);
} catch (NumberFormatException e) {
// 采用默認值
}
}
new HeartBeatsClient().connect(port, "127.0.0.1");
}

}
也稍微說明一下：
1）創建了ConnectionWatchdog對象，自然要實現handlers方法

2）初始化好bootstrap對象

3）4秒內沒有寫操作，進行心跳觸發，也就是IdleStateHandler這個方法

最後ConnectorIdleStateTrigger這個類

package com.lyncc.netty.idle;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;

@Sharable
public class ConnectorIdleStateTrigger extends ChannelInboundHandlerAdapter {

private static final ByteBuf HEARTBEAT_SEQUENCE = Unpooled.unreleasableBuffer(Unpooled.copiedBuffer("Heartbeat",
CharsetUtil.UTF_8));

@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
IdleState state = ((IdleStateEvent) evt).state();
if (state == IdleState.WRITER_IDLE) {
// write heartbeat to server
ctx.writeAndFlush(HEARTBEAT_SEQUENCE.duplicate());
}
} else {
super.userEventTriggered(ctx, evt);
}
}
}
HeartBeatClientHandler.java(不是重點)
package com.lyncc.netty.idle;

import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.ReferenceCountUtil;

import java.util.Date;

@Sharable
public class HeartBeatClientHandler extends ChannelInboundHandlerAdapter {


@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("激活時間是："+new Date());
System.out.println("HeartBeatClientHandler channelActive");
ctx.fireChannelActive();
}

@Override
public void channelInactive(ChannelHandlerContext ctx) throws Exception {
System.out.println("停止時間是："+new Date());
System.out.println("HeartBeatClientHandler channelInactive");
}


@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String message = (String) msg;
System.out.println(message);
if (message.equals("Heartbeat")) {
ctx.write("has read message from server");
ctx.flush();
}
ReferenceCountUtil.release(msg);
}
}


好了，到此為止，所有的代碼都貼完了，我們做一個簡單的測試，按照常理，如果不出任何狀況的話，客戶端4秒發送心跳，服務端5秒才驗證是不會斷連的，所以我們在啟動之後，關閉服務端，然後再次重啟服務端

首先啟動服務端，控制臺如下：

啟動客戶端，控制臺如下：

客戶端啟動之後，服務端的控制臺：

關閉服務端後，客戶端控制臺：

重啟啟動服務端：


---------------------
作者：BazingaLyncc
來源：CSDN
原文：https://blog.csdn.net/linuu/article/details/51509847
版權聲明：本文為博主原創文章，轉載請附上博文鏈接！

NETTY 心跳機制

相關推薦

NETTY 心跳機制

end value trap parseint acceptor blog channel 出發 每次 最近工作比較忙，但閑暇之余還是看了阿裏的馮家春（fengjiachun）的github上的開源代碼Jupiter，寫的RPC框架讓我感嘆人外有人，廢話不多說，下面的代碼全

淺析 Netty心跳機制

public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) { this((long)readerIdleTimeSeconds, (long)writerI

Netty 超時機制及心跳程序實現

timeunit pip demo ade reads unpooled 一個 apt group 本文介紹了 Netty 超時機制的原理，以及如何在連接閑置時發送一個心跳來維持連接。 Netty 超時機制的介紹 Netty 的超時類型 IdleState 主要分為： AL

Netty 實現心跳機制.md

ann 構造 red 通過 case 程序 evel exc 基本使用 netty 心跳機制示例，使用netty4，IdleStateHandler 實現。 本文假設你已經了解了Netty的使用，或者至少寫過netty的helloworld，知道了netty的基本使用。我們

netty 實現長連線，心跳機制，以及重連

實現的功能 心跳機制 and 長連線 and 重連機制 and 應用訊息傳輸： 這裡最關鍵的就是通過netty中的 IdleStateHandler 超時機制來實現心跳和重連 然後通過org.msgpack編碼器來實現跨平臺資料傳輸， 在這實現的功能就是通過Scanner來輸

Netty(一) SpringBoot 整合長連線心跳機制

前言 Netty 是一個高效能的 NIO 網路框架，本文基於 SpringBoot 以常見的心跳機制來認識 Netty。 最終能達到的效果： 客戶端每隔 N 秒檢測是否需要傳送心跳。 服務端也每隔 N 秒檢測是否需要傳送心跳。 服務端可以主動 push 訊息到客戶端。 基於 SpringBo

Netty學習(八)-Netty的心跳機制

1. 如何實現心跳機制一般實現心跳機制由兩種方式： TCP協議自帶的心跳機制來實現；在應用層來實現。但是TCP協議自帶的心跳機制系統預設是設定的是2小時的心跳頻率。它檢查不到機器斷電、網線拔出、防火牆這些斷線。而且邏輯層處理斷線可能也不是那麼好處理。另外該心跳機制是與TCP協議繫結的，那如果我們要是使用

netty自定義心跳機制

   在我的上一篇文章中已經介紹過了rts遊戲的基本架構，下面來看一下心跳協議的使用    在netty中重寫ChannelInboundHandlerAdapter的userEventTriggered方法可以實現心跳協議的檢測，寫起來也比較簡單，網上的demo很多，但是

聊聊心跳機制及netty心跳實現

　　我們在使用netty的時候會使用一個引數，ChannelOption.SO_KEEPALIVE為true, 設定好了之後再Linux系統才會對keepalive生效，但是linux裡邊需要配置幾個引數，tcp_keepalive_time, tcp_keepalive_invl, tcp_keepaliv

Netty 心跳服務之 IdleStateHandler 源碼分析

如何使用 讀寫 輸出 源碼 狀態 short 段落 今天 sys 前言：Netty 提供的心跳介紹 Netty 作為一個網絡框架，提供了諸多功能，比如我們之前說的編解碼，Netty 準備很多現成的編解碼器，同時，Netty 還為我們準備了網絡中，非常重要的一個服務----

心跳機制tcp keepalive的討論、應用及“斷網”、"斷電"檢測的C代碼實現(Windows環境下)

astar har 心跳 存在 假設 clu ali clean struct 版權聲明：本文為博主原創文章，轉載時請務必註明本文地址， 禁止用於任何商業用途， 否則會用法律維權。 https://blog.csdn.net/stpeace/article/details/

TCP長連線與短連線、心跳機制

1. TCP連線 當網路通訊時採用TCP協議時，在真正的讀寫操作之前，server與client之間必須建立一個連線，當讀寫操作完成後，雙方不再需要這個連線時它們可以釋放這個連線，連線的建立是需要三次握手的，而釋放則需要4次握手，所以說每個連線的建立都是需要資源消耗和時間消耗的。 經典的三

Netty 心跳相關(1)

無論是B/S還是C/S架構，如果你用的是長連線，那麼心跳是必不可少的。Netty提供了對心跳機制的天然支援，今天結合例子特地學習了一下。 首先，我們來設想一下何時需要傳送心跳。假設你做的是一款棋牌類小遊戲，那麼當玩家登陸游戲後肯定是先進入大廳，再選擇一張合適的桌子正式開始遊戲。此時玩

java 心跳機制

自己 查看 異常 定義 bsp 不存在 serve 時間段 部分 心跳機制：就是每隔幾分鐘發送一個固定信息給服務端，服務端收到後回復一個固定信息如果服務端幾分鐘內沒有收到客戶端信息則視客戶端斷開。 心跳包 心跳包就是在客戶端和服務器間定時通知對方自己狀態的一個自己

Mqtt ----心跳機制

心跳機制 　　Keep Alive指定連線最大空閒時間T，當客戶端檢測到連線空閒時間超過T時，必須向Broker傳送心跳報文PINGREQ，Broker收到心跳請求後返回心跳響應PINGRESP。若Broker超過1.5T時間沒收到心跳請求則斷開連線，並且投遞遺囑訊息到訂閱方；同樣，若客戶端超過一定時間仍沒

Dubbo服務合買平臺搭建出售釋出之服務暴露&心跳機制&服務註冊

Dubbo服務釋出 Dubbo合買平臺搭建出售 dsluntan.com Q:3393756370 VX:17061863513服務釋出影響流程的主要包括三個部分，依次是： 服務暴露 心跳 服務註冊 服務暴露是對外提供服務及暴露埠，以便消費端可以正常調通服務。心跳機制保證伺服器端及客戶

dubbo心跳機制 （1）

此文已由作者趙計剛授權網易雲社群釋出。 歡迎訪問網易雲社群，瞭解更多網易技術產品運營經驗。 dubbo的心跳機制： 目的：檢測provider與consumer之間的connection連線是不是還連線著，如果連線斷了，需要作出相應的處理。 原理： prov

dubbo心跳機制 （3）

此文已由作者趙計剛授權網易雲社群釋出。 歡迎訪問網易雲社群，瞭解更多網易技術產品運營經驗。 二、consumer端心跳機制              &nb

dubbo心跳機制 （2）

此文已由作者趙計剛授權網易雲社群釋出。 歡迎訪問網易雲社群，瞭解更多網易技術產品運營經驗。 來看一下HeaderExchangeServer.this.getChannels()：   1     public&nb

K8S 原始碼探祕 之 kubelet 同步 Node 狀態（kubelet 心跳機制分析）

一、引言        在 K8S 系統執行過程中，kubelet 需要定期向 API Server 上報節點執行狀態（也就是心跳訊息）        本文從原始碼角度分析下 kubelet 進行節點狀態上報