Netty(八) Netty心跳检测机制

1.什么是长链接和短链接

在HTTP/1.0中默认使用短连接。也就是说，客户端和服务器每进行一次HTTP操作，就建立一次连接，任务结束就中断连接。当客户端浏览器访问的某个HTML或其他类型的Web页中包含有其他的Web资源（如JavaScript文件、图像文件、CSS文件等），每遇到这样一个Web资源，浏览器就会重新建立一个HTTP会话。

而从HTTP/1.1起，默认使用长连接，用以保持连接特性。使用长连接的HTTP协议，会在响应头加入这行代码： Connection:keep-alive

在使用长连接的情况下，当一个网页打开完成后，客户端和服务器之间用于传输HTTP数据的TCP连接不会关闭，客户端再次访问这个服务器时，会继续使用这一条已经建立的连接。Keep-Alive不会永久保持连接，它有一个保持时间，可以在不同的服务器软件（如Apache）中设定这个时间。实现长连接需要客户端和服务端都支持长连接。 HTTP协议的长连接和短连接，实质上是TCP协议的长连接和短连接。

2. Netty心跳检测机制

2.1 所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 TCP 连接的有效性.我们需要明白大多数中间件的心跳都是客户端定时去往服务端发送心跳,服务端有个定时任务去检测来实现的.

在 Netty 中, 实现心跳机制的关键是 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 事件.

注：这三个参数默认的时间单位是秒。若需要指定其他时间单位，可以使用另一个构造方法：

要实现Netty服务端心跳检测机制需要在服务器端的ChannelInitializer中加入如下的代码：

当时间为0代表不会触发对应的事件

 pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));

下面代码逻辑是客户端在,随机时间往服务端发送一段'Heartbeat Packet',而服务端的逻辑是,添加了心跳的IdleStateHandler,如果3次以上内没感知到读事件,则会强行关闭channel

//服务端代码
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>() {@Overrideprotected 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, TimeUnit.SECONDS));pipeline.addLast(new HeartBeatHandler());}});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();}}
}//服务端处理handler
public class HeartBeatServerHandler extends SimpleChannelInboundHandler<String> {int readIdleTimes = 0;@Overrideprotected 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(" 其他信息处理 ... ");}}@Overridepublic void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {IdleStateEvent event = (IdleStateEvent) evt;String eventType = null;switch (event.state()) {case READER_IDLE:eventType = "读空闲";readIdleTimes++; // 读空闲的计数加1break;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();}}@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");}
}//客户端代码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>() {@Overrideprotected 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> {@Overrideprotected 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();}}}
}

运行结果:

2.2 IdleStateHandler源码实现

类结构表明他既是InboundHandler也是OutboudHandler,所以,数据读和写的时候都会触发他里面的方法

先看下IdleStateHandler中的channelRead方法：并没有什么特别,就是调用这个方法的时候调用了ctx.fireChannelRead(msg);该方法只是进行了透传，不做任何业务逻辑处理，让channelPipe中的下一个handler处理channelRead方法

  public class IdleStateHandler extends ChannelDuplexHandler {private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1);@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {reading = true;firstReaderIdleEvent = firstAllIdleEvent = true;}ctx.fireChannelRead(msg);}
}

我们再看看channelActive方法：这里有个initialize的方法，这是IdleStateHandler的精髓，我们看到了schedule肯定就会想到了定时任务

public class IdleStateHandler extends ChannelDuplexHandler {
...@Overridepublic 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);}private void initialize(ChannelHandlerContext ctx) {// Avoid the case where destroy() is called before scheduling timeouts.// See: https://github.com/netty/netty/issues/143switch (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);}}}

我们可以看到上面触发的定时任务肯定会触发下面这个run方法,他if和else里面逻辑是一样的触发下一个定时任务,但是else里面还多了一个channelIdel方法,他的作用是触发下一个Handler的userEventTriggered方法
nextDelay -= ticksInNanos() - Math.max(lastReadTime, lastWriteTime);这句话的意思是nextDelay =3秒 -(当前时间-最后一次读或者写的时间),3秒是我们传的参数3,如果小于0,就代表超时了,会触发下一个handler的userEventTriggered方法
为什么不用固定时间的定时任务去做?因为如果没超时的时候,下一个定时任务的时间并不是固定的,是nextDelay,假如我们的nextDelay算出来是1(1=3-2),证明我们2S之前来了个心跳,那么下一次执行任务就是1S之后.假如超时了,下一次心跳检测任务就是3S之后执行.

    private final class AllIdleTimeoutTask extends AbstractIdleTask {AllIdleTimeoutTask(ChannelHandlerContext ctx) {super(ctx);}@Overrideprotected void run(ChannelHandlerContext ctx) {long nextDelay = allIdleTimeNanos;if (!reading) {nextDelay -= ticksInNanos() - Math.max(lastReadTime, lastWriteTime);}if (nextDelay <= 0) {// Both reader and writer are idle - set a new timeout and// notify the callback.allIdleTimeout = schedule(ctx, this, allIdleTimeNanos, TimeUnit.NANOSECONDS);boolean first = firstAllIdleEvent;firstAllIdleEvent = false;try {if (hasOutputChanged(ctx, first)) {return;}IdleStateEvent event = newIdleStateEvent(IdleState.ALL_IDLE, first);channelIdle(ctx, event);} catch (Throwable t) {ctx.fireExceptionCaught(t);}} else {// Either read or write occurred before the timeout - set a new// timeout with shorter delay.allIdleTimeout = 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 {ctx.fireUserEventTriggered(evt);}

每次读完客户端的数据会更新LastReadTime