Netty4实现心跳检测及IdleStateHandler源码分析

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
/**
 * 服务端心跳检测
 */
public class HeartBeatHandler extends ChannelInboundHandlerAdapter {

	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

		if(evt instanceof IdleStateEvent){
			IdleStateEvent ise = (IdleStateEvent)evt;
			
			// 服务端读空闲
			if(ise.state().equals(IdleState.READER_IDLE)){
				ctx.writeAndFlush("client reader idle, channel will close");
				ctx.channel().close();
				
			// 服务端写空闲
			}else if (ise.state().equals(IdleState.WRITER_IDLE)){
				ctx.write("pong message");
				
			// 服务端读写空闲
			}else if (ise.state().equals(IdleState.ALL_IDLE)){
				ctx.channel().close();
			}else{
				// DO NOTHING
			}
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
public class Server {

	public static void main(String[] args) {
		//服务类
		ServerBootstrap bootstrap = new ServerBootstrap();
		
		//boss和worker
		EventLoopGroup boss = new NioEventLoopGroup();
		EventLoopGroup worker = new NioEventLoopGroup();
		
		try {
			//设置线程池
			bootstrap.group(boss, worker);
			
			//设置socket工厂
			bootstrap.channel(NioServerSocketChannel.class);
			
			//设置管道工厂
			bootstrap.childHandler(new ChannelInitializer<Channel>() {

               
				@Override
				protected void initChannel(Channel ch) throws Exception {
					ch.pipeline().addLast(new StringEncoder());
                     // 主要在这里，先创建一个IdleStateHandler，10秒内没有读写则判定为空闲
					ch.pipeline().addLast(new IdleStateHandler(10, 10, 10, TimeUnit.SECONDS));
                    // 然后执行心跳检测
					ch.pipeline().addLast(new HeartBeatHandler());
				}
			});
			
			//设置参数，TCP参数
			bootstrap.option(ChannelOption.SO_BACKLOG, 2048);//serverSocketchannel的设置，链接缓冲池的大小
			bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);//socketchannel的设置,维持链接的活跃，清除死链接
			bootstrap.childOption(ChannelOption.TCP_NODELAY, true);//socketchannel的设置,关闭延迟发送
			
			//绑定端口
			ChannelFuture future = bootstrap.bind(8088).sync();
			
			System.out.println("server start...");
			//等待服务端关闭
			future.channel().closeFuture().sync();
			
		} catch (Exception e) {
			e.printStackTrace();
		} finally{
			//释放资源
			boss.shutdownGracefully();
			worker.shutdownGracefully();
		}
	}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
public class IdleStateHandler extends ChannelDuplexHandler {
    
    // 对象初始化的时候初始化好这些数据
    private final long readerIdleTimeNanos;
    private final long writerIdleTimeNanos;
    private final long allIdleTimeNanos;

    // 通过构造函数可知
    public IdleStateHandler(
            long readerIdleTime, long writerIdleTime, long allIdleTime,
            TimeUnit unit) {
        if (unit == null) {
            throw new NullPointerException("unit");
        }

        if (readerIdleTime <= 0) {
            readerIdleTimeNanos = 0;
        } else {
            readerIdleTimeNanos = Math.max(unit.toNanos(readerIdleTime), MIN_TIMEOUT_NANOS);
        }
        if (writerIdleTime <= 0) {
            writerIdleTimeNanos = 0;
        } else {
            writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
        }
        if (allIdleTime <= 0) {
            allIdleTimeNanos = 0;
        } else {
            allIdleTimeNanos = Math.max(unit.toNanos(allIdleTime), MIN_TIMEOUT_NANOS);
        }
    }
    
    // 在channelRegistered/channelActive的时候调用初始化方法initialize方法
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        // Initialize early if channel is active already.
        if (ctx.channel().isActive()) {
            initialize(ctx);
        }
        super.channelRegistered(ctx);
    }

    @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);
    }
    
    // 监听到客户端写数据的时候，更新lastReadTime
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        lastReadTime = System.nanoTime();
        firstReaderIdleEvent = firstAllIdleEvent = true;
        ctx.fireChannelRead(msg);
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
private void initialize(ChannelHandlerContext ctx) {
    // Avoid the case where destroy() is called before scheduling timeouts.
    // See: https://github.com/netty/netty/issues/143
    
    // private volatile int state; // 0 - none, 1 - initialized, 2 - destroyed
    switch (state) {
        case 1:
        case 2:
            return;
    }

    state = 1;

    // 1.获取线程池
    EventExecutor loop = ctx.executor();

    // 2.初始化最后一次读时间和最后一次写时间为当前时间
    lastReadTime = lastWriteTime = System.nanoTime();
    
    // 3.根据用户设置的读空闲时间启动一个定时任务，读空闲时间为频率执行
    if (readerIdleTimeNanos > 0) {
        readerIdleTimeout = loop.schedule(
            new ReaderIdleTimeoutTask(ctx),
            readerIdleTimeNanos, TimeUnit.NANOSECONDS);
    }
    
    // 4.根据用户设置的写空闲时间启动一个定时任务，写空闲时间为频率执行
    if (writerIdleTimeNanos > 0) {
        writerIdleTimeout = loop.schedule(
            new WriterIdleTimeoutTask(ctx),
            writerIdleTimeNanos, TimeUnit.NANOSECONDS);
    }
    
    // 5.根据用户设置的读写空闲时间启动一个定时任务，读写空闲时间为频率执行
    if (allIdleTimeNanos > 0) {
        allIdleTimeout = loop.schedule(
            new AllIdleTimeoutTask(ctx),
            allIdleTimeNanos, TimeUnit.NANOSECONDS);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
private final class ReaderIdleTimeoutTask implements Runnable {

    private final ChannelHandlerContext ctx;

    ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
        this.ctx = ctx;
    }

    @Override
    public void run() {
        if (!ctx.channel().isOpen()) {
            return;
        }

        // 1.获取当前时间和最后一次读时间
        long currentTime = System.nanoTime();
        long lastReadTime = IdleStateHandler.this.lastReadTime;
        
        // nextDelay即为比较用户设置的读空闲时间和 当前时间-最后一次时间
        long nextDelay = readerIdleTimeNanos - (currentTime - lastReadTime);
        
        // 2.如果nextDelay<=0，则说明客户端已经在超过读空闲时间内没有写入数据了
        if (nextDelay <= 0) {
            // 2.1 重新定义readerIdleTimeout schedule，与initialize方法设置的相同，继续执行定时任务
            readerIdleTimeout =
                ctx.executor().schedule(this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
            try {
                // 2.2 event = new IdleStateEvent(IdleState.READER_IDLE, true)，将event设置为读空闲
                IdleStateEvent event;
                if (firstReaderIdleEvent) {
                    firstReaderIdleEvent = false;
                    event = IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT;
                } else {
                    event = IdleStateEvent.READER_IDLE_STATE_EVENT;
                }
                // 2.3 channelIdle的主要工作就是将evt传输给下一个Handler
                channelIdle(ctx, event);
            } catch (Throwable t) {
                ctx.fireExceptionCaught(t);
            }
            
        // 3.如果nextDelay>0，则说明客户端在规定时间内已经写入数据了，
        } else {
            // 3.1 重新定义readerIdleTimeout schedule，以nextDelay为执行频率
            readerIdleTimeout = ctx.executor().schedule(this, nextDelay, TimeUnit.NANOSECONDS);
        }
    }
}