一. 概念引入
1. 心跳
在TCP長連接配接中，用戶端和服務端之間定期收發的一種特殊的資料包稱為“心跳包”，用以通知和确認對方都還線上，以確定TCP連接配接的有效性
2. 心跳的必要性
- 用戶端程式崩潰、或者網絡斷開等原因，單方面釋放了TCP連接配接
- TCP連接配接被防火牆幹掉
計算機與計算機之間互相是有防火牆的，而這個防火牆随時可以做到一個政策，随時可以斷開socket連接配接，而斷開的時候可能不會進行四次揮手，服務端或者用戶端沒有收到連接配接斷開的消息，此時會認為連接配接還可用，随時還想發送資料，發送的時候才知道連接配接不可用。
一般來說，正是因為如上這些導緻TCP長連接配接斷開的不确定因素（用戶端的因素比較常見），才需要“心跳包”來确認雙方是否線上。服務端得知用戶端單方面釋放掉TCP連接配接，服務端會及時釋放相應的記憶體資源。而用戶端知道TCP連接配接被斷開了，根據需要來采取措施是否需要重連
二. 帶有注釋的demo
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
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.timeout.IdleStateHandler;
public class HeartBeatServer {
public static void main(String[] args) throws Exception {
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup worker = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(boss, worker)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
// IdleStateHandler的readerIdleTime參數指定超過3秒還沒收到用戶端的連接配接，
// 會觸發IdleStateEvent事件并且交給下一個handler處理，下一個handler必須
// 實作userEventTriggered方法處理對應事件
pipeline.addLast(new IdleStateHandler(3, 0, 0));
pipeline.addLast(new HeartBeatServerHandler());
}
});
System.out.println("netty server start。。");
ChannelFuture future = bootstrap.bind(9000).sync();
future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
}
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
public class HeartBeatServerHandler extends SimpleChannelInboundHandler<String> {
// 讀逾時的次數
int readIdleTimes;
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
System.out.println(" ====== > [server] message received : " + s);
if ("Heartbeat Packet".equals(s)) {
// 如果是心跳包，給用戶端發送一個響應 "ok"
ctx.channel().writeAndFlush("ok");
} else {
System.out.println("其他資訊處理 ... ");
}
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
IdleStateEvent event = (IdleStateEvent) evt;
String eventType = null;
switch (event.state()) {
case READER_IDLE:
eventType = "讀空閑";
readIdleTimes++; // 讀空閑的計數加1
break;
case WRITER_IDLE:
eventType = "寫空閑";
// 不處理
break;
case ALL_IDLE:
eventType = "讀寫空閑";
// 不處理
break;
}
System.out.println(ctx.channel().remoteAddress() + "逾時事件：" + eventType);
if (readIdleTimes > 3) {
// 即時超過3次，也不一定是TCP連接配接真的挂掉了
// 也有可能是傳輸過程中網絡擁堵，進而造成服務端在誤判為TCP連接配接斷開了
System.out.println(" [server]讀空閑超過3次，關閉連接配接，釋放更多資源");
// 關閉通道之前，先嘗試發送一個關閉資訊
ctx.channel().writeAndFlush("idle close");
ctx.channel().close();
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
// 通道就緒之後，初始化讀逾時的次數為0
readIdleTimes = 0;
}
}
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.util.Random;
public class HeartBeatClient {
public static void main(String[] args) throws Exception {
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast(new HeartBeatClientHandler());
}
});
System.out.println("netty client start。。");
Channel channel = bootstrap.connect("127.0.0.1", 9000).sync().channel();
String text = "Heartbeat Packet";
Random random = new Random();
while (channel.isActive()) {
int num = random.nextInt(10);
Thread.sleep(num * 1000);
channel.writeAndFlush(text); // 模拟發送心跳包
}
} catch (Exception e) {
e.printStackTrace();
} finally {
eventLoopGroup.shutdownGracefully();
}
}
static class HeartBeatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println("client received :" + msg);
if (msg != null && msg.equals("idle close")) {
System.out.println(" 服務端關閉連接配接，用戶端也關閉");
ctx.channel().closeFuture();
}
}
}
}
三. Netty的心跳機制的關鍵源碼分析
1. IdleStateHandler
IdleStateHandler
public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
}
- readerIdleTimeSeconds: 讀逾時. 即當在指定的時間間隔内沒有從 Channel 讀取到資料時, 會觸發一個 READER_IDLE 的 IdleStateEvent 事件
- writerIdleTimeSeconds: 寫逾時. 即當在指定的時間間隔内沒有資料寫入到 Channel 時, 會觸發一個 WRITER_IDLE 的 IdleStateEvent 事件
- allIdleTimeSeconds: 讀/寫逾時. 即當在指定的時間間隔内沒有讀或寫操作時, 會觸發一個 ALL_IDLE 的 IdleStateEvent 事件
2. 大膽猜想
在此之前，先來大膽猜想一波。
在demo中，設定了讀逾時參數為3秒。那麼Netty會每隔3秒就會判斷是否能夠從通道中讀取到資料（可能是心跳包，也可能是業務資料）。很顯然，隻需要一個定時器。那麼，很容易才想到定時器的初始化必定是在通道就緒之後。是以我們定位
IdleStateHandler
中的
channelActive()
方法，發現了核心的
initialize()
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
// This method will be invoked only if this handler was added
// before channelActive() event is fired. If a user adds this handler
// after the channelActive() event, initialize() will be called by beforeAdd().
initialize(ctx); // 關鍵！！！
super.channelActive(ctx);
}
3. IdleStateHandler
initialize()
IdleStateHandler
initialize()
從源碼中可以看出，Netty并沒有直接使用定時器，而是使用了 “延時任務”（遞歸調用） 來實作更加靈活的定時器
private void initialize(ChannelHandlerContext ctx) {
// Avoid the case where destroy() is called before scheduling timeouts.
// See: https://github.com/netty/netty/issues/143
switch (state) {
case 1:
case 2:
return;
}
state = 1;
initOutputChanged(ctx);
lastReadTime = lastWriteTime = ticksInNanos();
if (readerIdleTimeNanos > 0) {
// 關鍵！！！
readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
readerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (writerIdleTimeNanos > 0) {
writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
writerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (allIdleTimeNanos > 0) {
allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
allIdleTimeNanos, TimeUnit.NANOSECONDS);
}
}
/**
* This method is visible for testing!
*/
ScheduledFuture<?> schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
return ctx.executor().schedule(task, delay, unit);
}
4. 檢視讀逾時的定時任務 ReaderIdleTimeoutTask
ReaderIdleTimeoutTask
private final class ReaderIdleTimeoutTask extends AbstractIdleTask {
ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
super(ctx);
}
@Override
protected void run(ChannelHandlerContext ctx) {
// nextDelay：顧名思義，下一次任務應該在延時多少秒之後執行
// lastReadTime：顧名思義，上一次從通道中讀取到資料的時間
long nextDelay = readerIdleTimeNanos;
if (!reading) {
// 等價于 nextDelay = nextDelay - (ticksInNanos() - lastReadTime);
// 以demo的讀逾時為3秒為例
/*
假如(ticksInNanos()-lastReadTime)為2，說明此時讀到資料，距離上次讀取到資料為2秒。
小于傳參的3秒，換言之，就是下一個3秒的“節點”還沒有到。
那麼下一次的任務應該在3-2=1秒之後執行。是以新的nextDelay就為1
*/
/*
假如(ticksInNanos()-lastReadTime)為4，說明此時讀到資料，距離上次讀取到資料為4秒。
大于傳參的3秒，換言之下一個3秒的“節點”已經過了。
此時按照下面代碼算出來的nextDelay就是-1，此時需要重新開啟一個3秒的延時任務
*/
nextDelay -= ticksInNanos() - lastReadTime;
}
if (nextDelay <= 0) {
// Reader is idle - set a new timeout and notify the callback.
// 重新開啟一個3秒延時任務。延時3秒後執行任務
readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
boolean first = firstReaderIdleEvent;
firstReaderIdleEvent = false;
try {
IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
// 調用下一個管道中IdleStateHandler的下一個handler的userEventTriggered方法來處理讀逾時事件
channelIdle(ctx, event);
} catch (Throwable t) {
ctx.fireExceptionCaught(t);
}
} else {
// Read occurred before the timeout - set a new timeout with shorter delay.
// 延時nextDelay秒後執行任務
readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
}
}
}
/**
* Is called when an {@link IdleStateEvent} should be fired. This implementation calls
* {@link ChannelHandlerContext#fireUserEventTriggered(Object)}.
*/
protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
// 凡是fireXxx()方法，都是調用管道中目前handler的下一個handler的Xxx()方法來繼續消費資料
ctx.fireUserEventTriggered(evt);
}
![Netty之EventLoop[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)