Netty心跳检测机制

Netty心跳检测机制

Netty心跳检测机制

• 1、Netty心跳机制简介
• • 1.1、什么是心跳机制
  • 1.2、为什么要有Netty心跳机制？
• 2、实现 Netty 的心跳机制
• • 2.1、IdleStateHandler的超时逻辑判断分析
  • 2.2、重点代码
  • • 2.2.1、重试重点源码剖析
    • 2.2.2、自定义处理类

1、Netty心跳机制简介 1.1、什么是心跳机制

所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.

1.2、为什么要有Netty心跳机制？

因为如果服务端不知道某个客户端已经宕机了或者出现其他问题客户端服务器已经不能用了，然而现在服务端还保持着和这个客户端的连接，会浪费资源。所以有了心跳机制之后，服务端就知道客户端在某一时刻还存活着，就可以相互之间通信，当客户端挂了之后，服务端也会立马感知到并释放相应的资源。

2、实现 Netty 的心跳机制

在 Netty 中, 实现心跳机制的关键是 IdleStateHandler, 看下它的构造器：
在 ChannelHandler 直接添加下面这个类即可。

public IdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime, TimeUnit unit) {
    this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
}

2.1、IdleStateHandler的超时逻辑判断分析

2.2、重点代码 2.2.1、重试重点源码剖析

protected void run(ChannelHandlerContext ctx) {
    long nextDelay = IdleStateHandler.this.readerIdleTimeNanos;
    if (!IdleStateHandler.this.reading) {
    	// IdleStateHandler.this.ticksInNanos() - IdleStateHandler.this.lastReadTime 是当前时间减去上一次读取数据的时间
    	// nextDelay 是设置的超时时间
        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);
            // 这里面会调用下一个handler的fireUserEventTriggered方法
            IdleStateHandler.this.channelIdle(ctx, event);
        } catch (Throwable var6) {
            ctx.fireExceptionCaught(var6);
        }
    } 
	// 没有超时，将和超时的时间差补齐再次执行，形成一个定时任务
	else {
        IdleStateHandler.this.readerIdleTimeout = IdleStateHandler.this.schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
    }

}

2.2.2、自定义处理类

这个得实现上面调用的 fireUserEventTriggered 方法

public class HeartBeatServerHandler extends SimpleChannelInboundHandler {

    int readIdleTimes = 0;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
        System.out.println(" ====== > [server] message received : " + s);
        if ("Heartbeat Packet".equals(s)) {
            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) {
            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 ===");
    }
}