Netty的心跳检测机制源码阅读
一.原理与源码阅读
Netty4.0提供了一个类，名为IdleStateHandler，这个类可以对三种类型的心跳检测。
....
public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
}
....
前三个的参数解释如下：
1）readerIdleTime：为读超时时间（即测试端一定时间内未接受到被测试端消息）
2）writerIdleTime：为写超时时间（即测试端一定时间内向被测试端发送消息）
3）allIdleTime：所有类型的超时时间
idleState示例
我们在pipeLine中加入了IdleSateHandler，第一个参数是readerIdleTimeSeconds，数值为5，那么，在服务器端会每隔5秒来检查一下channelRead方法被调用的情况，如果在5秒内pipeLine中的channelRead方法都没有被触发，就会调用userEventTriggered方法。
下面我们看一下源码是怎么调用的...
public void channelActive(ChannelHandlerContext ctx) throws Exception {
this.initialize(ctx);
super.channelActive(ctx);
}
首先是channelActive，在channel**时，会触发初始化,下面是初始化的方法
private void initialize(ChannelHandlerContext ctx) {
switch(this.state) {
case 1:
case 2:
return;
default:
this.state = 1;
this.initOutputChanged(ctx);
this.lastReadTime = this.lastWriteTime = this.ticksInNanos();
if (this.readerIdleTimeNanos > 0L) {
//注意这里
this.readerIdleTimeout = this.schedule(ctx, new IdleStateHandler.ReaderIdleTimeoutTask(ctx), this.readerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (this.writerIdleTimeNanos > 0L) {
this.writerIdleTimeout = this.schedule(ctx, new IdleStateHandler.WriterIdleTimeoutTask(ctx), this.writerIdleTimeNanos, TimeUnit.NANOSECONDS);
}
if (this.allIdleTimeNanos > 0L) {
this.allIdleTimeout = this.schedule(ctx, new IdleStateHandler.AllIdleTimeoutTask(ctx), this.allIdleTimeNanos, TimeUnit.NANOSECONDS);
}
}
}
这里会根据传入的参数来添加一些定时任务，看一下ReaderIdleTimeoutTask的源码
private final class ReaderIdleTimeoutTask extends IdleStateHandler.AbstractIdleTask {
ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
super(ctx);
}
protected void run(ChannelHandlerContext ctx) {
long nextDelay = IdleStateHandler.this.readerIdleTimeNanos;
if (!IdleStateHandler.this.reading) {
//注意这里，用当前时间减去最后一次channelRead方法调用的时间，与nextDeay相减
nextDelay -= IdleStateHandler.this.ticksInNanos() - IdleStateHandler.this.lastReadTime;
}
if (nextDelay <= 0L) {
//如果超时
IdleStateHandler.this.readerIdleTimeout = IdleStateHandler.this.schedule(ctx, this, IdleStateHandler.this.readerIdleTimeNanos, TimeUnit.NANOSECONDS);
boolean first = IdleStateHandler.this.firstReaderIdleEvent;
IdleStateHandler.this.firstReaderIdleEvent = false;
try {
//生成一个事件
IdleStateEvent event = IdleStateHandler.this.newIdleStateEvent(IdleState.READER_IDLE, first);
//调用channelIdle方法
IdleStateHandler.this.channelIdle(ctx, event);
} catch (Throwable var6) {
ctx.fireExceptionCaught(var6);
}
} else {
IdleStateHandler.this.readerIdleTimeout = IdleStateHandler.this.schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
}
}
}
可以看到，用当前时间减去最后一次channelRead方法调用的时间，假如这个结果是6s，说明最后一次调用channelRead已经是6s之前的事情了，你设置的是5s，那么nextDelay则为-1，说明超时了，那么会触发channelIdle方法，下面看一下channelIdle的源码
protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
ctx.fireUserEventTriggered(evt);
}
可以看到，会触发fireUserEventTriggered方法的执行，调用pipeLine中的UserEventTriggered方法。
如果没有超时，则会循环定时检测，除非你将IdleStateHandler移除Pipeline。
这就是IdleStateHandler的基本原理了。
二.代码示例
1.服务器端
AcceptorIdleStateTrigger 中对userEventTriggered方法重写，主要作用是，当服务器长时间没有接受到客户端数据，会抛出相应的异常，然后根据异常进行相应的处理，这里我后面的handler会捕获异常，然后将Channel关闭。
@ChannelHandler.Sharable
public class AcceptorIdleStateTrigger extends ChannelInboundHandlerAdapter {
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent stateEvent = (IdleStateEvent) evt;
IdleState state = stateEvent.state();
if (state==IdleState.READER_IDLE){
throw new Exception("客户端空闲时间长，自动断开连接");
}
}else {
super.userEventTriggered(ctx,evt);
}
}
}
HeartServerHandler 自定义的处理器
public class HeartServerHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.out.println("Server Active");
}
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
Message message= (Message) msg;
if (message.getType()==(byte)Integer.parseInt("CF", 16)){
System.out.println(ctx.channel().remoteAddress()+">>>"+message.getType()+":"+message.getRequestId()+":"+message.getBody());
ctx.writeAndFlush(new Message(message.getType(),message.getRequestId(),"server time->"+new Date()));
}else if (message.getType()==(byte)Integer.parseInt("AF", 16)){
System.out.println(ctx.channel().remoteAddress()+">>>>>>>>>>>>>>>>"+message.getBody());
}
ctx.fireChannelRead(msg);
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
System.out.println("已断开客户端"+ctx.channel().remoteAddress());
ctx.channel().close();
}
}
HeartServer 服务端代码
public class HeartServer {
public static void main(String[] args){
HeartServer server = new HeartServer();
server.init(8080);
}
public void init (int port){
NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
NioEventLoopGroup workerGroup = new NioEventLoopGroup(10);
ServerBootstrap serverBootstrap = new ServerBootstrap();
AcceptorIdleStateTrigger idleStateTrigger = new AcceptorIdleStateTrigger();
serverBootstrap.option(ChannelOption.SO_BACKLOG, 128).
group(bossGroup,workerGroup).
channel(NioServerSocketChannel.class).
childOption(ChannelOption.SO_KEEPALIVE, true).
childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new IdleStateHandler(5,0,0));
pipeline.addLast(new MessageDecoder(Integer.MAX_VALUE,9,4,0,0,false));
pipeline.addLast(new MessageEncoder());
pipeline.addLast(idleStateTrigger);
pipeline.addLast(new HeartServerHandler());
}
});
try {
ChannelFuture future = serverBootstrap.bind(port).sync();
System.out.println("server start ...");
//服务器同步连接断开时,这句代码才会往下执行
future.channel().closeFuture().sync();
System.out.println("server stop ...");
} catch (InterruptedException e) {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
e.printStackTrace();
}
}
}
2.客户端
客户端长时间没有写数据，会触发下面的userEventTriggered方法，然后根据idle状态发送心跳包，保证客户端和服务器端的连接不会断开。
@ChannelHandler.Sharable
public class ConnectorIdleStateTrigger extends ChannelInboundHandlerAdapter {
private final Message heartMsg = new Message((byte)Integer.parseInt("AF",16), System.currentTimeMillis(),"Heartbeat");
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent){
IdleStateEvent stateEvent = (IdleStateEvent) evt;
IdleState state = stateEvent.state();
if (state == IdleState.WRITER_IDLE){
ctx.writeAndFlush(heartMsg);
}
}else{
super.userEventTriggered(ctx, evt);
}
}
}
心跳机制实现.png