Netty学习篇④-心跳机制及断线重连

心跳检测

1. 前言

   客户端和服务端的连接属于socket连接，也属于长连接，往往会存在客户端在连接了服务端之后就没有任何操作了，但还是占用了一个连接；当越来越多类似的客户端出现就会浪费很多连接，netty中可以通过心跳检测来找出一定程度（自定义规则判断哪些连接是无效链接）的无效链接并断开连接，保存真正活跃的连接。

2. 什么叫心跳检测

   我理解的心跳检测应该是客户端/服务端定时发送一个数据包给服务端/客户端，检测对方是否有响应；
   如果是存活的连接，在一定的时间内应该会收到响应回来的数据包；
   如果在一定时间内还是收不到接收方的响应的话，就可以当做是挂机，可以断开此连接；
   如果检测到了掉线之后还可以进行重连；

3. 心跳检测的实现

   • TCP自带心跳检测，协议层采用Keeplive机制默认2小时频率触发一次检测，但是它存在缺陷:检测不出网线拔出、防火墙、使用起来不灵活、依赖操作系统等
   • Netty可以通过IdleStateHandler来实现心跳检测，使用起来也非常方便清晰

4. IdleStateHandler原理

   idleStateHandler在通道注册之后会开启一个定时任务，定时去检测通道中后续是否还有进行数据传输，如果在规定的时间内没有进行数据传输则会触发对应的超时事件，使用者可以根据对应的事件自定义规则来判别当前连接是否是活跃，是否需要关闭连接等来进行操作。
   一般idleStateHandler触发的事件IdleStateEvent会在心跳handler中的userEventTriggered方法中捕获到对应的超时事件。

   IdleStateHandler的继承关系：通过ChannelDuplexHandler类继承ChannelInboundHandler和实现ChannelOutboundHandler来实现对入站和出站的重写和监控

   

5. 源码分析

   • IdleStateHandler初始化：为0代表不监控

     /**
     * @observeOutput 观察输出
     * @readerIdleTime 读超时时间 自定义时间内检测Channel通道有没有读取到数据，为0代表不监控
     * @writerIdleTime 写超时时间 自定义时间内检测Channel通道有没有write数据，为0代表不监控
     * @allIdleTime 总超时时间 自定义时间内检测Channel通道有没有读/写数据，为0代表不监控
     * @unit 时间单位
     */
     public IdleStateHandler(boolean observeOutput,
                 long readerIdleTime, long writerIdleTime, long allIdleTime,
                 TimeUnit unit) {
             if (unit == null) {
                 throw new NullPointerException("unit");
             }
     
             this.observeOutput = observeOutput;
     
            // 初始化读取空闲时间，最小值为0
             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);
             }
         }

   • IdleStateHandler和channel通道的关联：通过类图可以得知，idleStateHandler可以重写入站和出站的方法，不过只是通过channelRead和write方法来记录阅读的时间等不做其他操作

     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         // 初始化检测器，开启定时任务
         initialize(ctx);
         super.channelActive(ctx);
     }
     
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         // 在通道非活跃状态的时候销毁定时任务
         destroy();
         super.channelInactive(ctx);
     }
     
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         // 如果设置的读超时时间大于0则设置是否读操作为true
         if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
             reading = true;
             firstReaderIdleEvent = firstAllIdleEvent = true;
         }
         // 记录时间和标识标志之后就直接fire当前read到下一个ChannelHandler处理类中
         ctx.fireChannelRead(msg);
     }
     
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
         // 当读取完毕时，设置是否正在读取为false，设置最后读取时间为系统当前时间
         if ((readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) && reading) {
             lastReadTime = ticksInNanos();
             reading = false;
         }
         // 同样直接fire掉，跳入到下一个handler中
         ctx.fireChannelReadComplete();
     }
     
     // idleStateHandler重写的write方法
     @Override
     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
         // Allow writing with void promise if handler is only configured for read timeout events.
         if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
             ctx.write(msg, promise.unvoid()).addListener(writeListener);
         } else {
             ctx.write(msg, promise);
         }
     }
     
     // ChannelOutboundHandler提供的write接口方法
     /**
         * Called once a write operation is made. The write operation will write the messages through the
          * {@link ChannelPipeline}. Those are then ready to be flushed to the actual {@link Channel} once
          * {@link Channel#flush()} is called
          * 执行一次写操作；写操作通过ChannelPipeline来传输信息；最后通过channel的flush()方法来刷新
          *
          * @param ctx               the {@link ChannelHandlerContext} for which the write operation is made 实际写操作者
          * @param msg               the message to write 写的消息
          * @param promise           the {@link ChannelPromise} to notify once the operation completes 在操作完成时立即通知（类似future异步通知）
          * @throws Exception        thrown if an error occurs
          */
         void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;

   • 定时器初始化/销毁：将使用者自定义的超时时间设置为延迟定时任务

     初始化：
     
     private void initialize(ChannelHandlerContext ctx) {
             // Avoid the case where destroy() is called before scheduling timeouts.
             // See: https://github.com/netty/netty/issues/143
            // state; // 0 - none, 1 - 初始化, 2 - 销毁
             switch (state) {
             case 1:
             case 2:
                 return;
             }
     
             state = 1;
             initOutputChanged(ctx);
            // 最后读取时间
             lastReadTime = lastWriteTime = ticksInNanos();
            // 如果设置的空闲时间大于0则开启定时任务进行监控
             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);
             }
         }

     销毁：
     
     /**
     * @readerIdleTimeout ==> ScheduledFuture类
     * @writerIdleTimeout ==> ScheduledFuture类
     */
     private void destroy() {
            // 设置销毁状态
             state = 2;
            // 销毁线程
            // ScheduledFuture
             if (readerIdleTimeout != null) {
                 readerIdleTimeout.cancel(false);
                 readerIdleTimeout = null;
             }
             if (writerIdleTimeout != null) {
                 writerIdleTimeout.cancel(false);
                 writerIdleTimeout = null;
             }
             if (allIdleTimeout != null) {
                 allIdleTimeout.cancel(false);
                 allIdleTimeout = null;
             }
         }

   • 开启定时任务

     ScheduledFuture<?> schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
             return ctx.executor().schedule(task, delay, unit);
         }

   • 读/写定时任务：定时任务启动的时候，通过设置的超时时间和上一次触发channelRead的时间进行相减比较来判断是否超时了

     // 仅分析读取超时时间定时任务，写超时差不多就是触发的时间不一样，比对的变量换成了设置的写超时时间
     private final class ReaderIdleTimeoutTask extends AbstractIdleTask {
     
             ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
                 super(ctx);
             }
     
             @Override
             protected void run(ChannelHandlerContext ctx) {
                 // readerIdleTimeNanos：初始化IdleStateHandler设置的读取超时时间
                 long nextDelay = readerIdleTimeNanos;
                 // 如果没有任何读取操作
                 if (!reading) {
                     // 判断是否有超时
                     // nextDelay = nextDelay-（ticksInNanos() - lastReadTime）
                     // 即设置的超时时间减去距离上一次读取的时间
                     nextDelay -= ticksInNanos() - lastReadTime;
                 }
                // 如果小于等于0 则触发读取超时事件,设置新的延迟时间
                 if (nextDelay <= 0) {
                     // Reader is idle - set a new timeout and notify the callback.
                     readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
                    // 标记为第一次
                     boolean first = firstReaderIdleEvent;
                     // 设置成非第一次
                     firstReaderIdleEvent = false;
     
                     try {
                         IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
                         channelIdle(ctx, event);
                     } catch (Throwable t) {
                         ctx.fireExceptionCaught(t);
                     }
                 } else {
                     // Read occurred before the timeout - set a new timeout with shorter delay.
                     // 超时的时候发生的读取事件，则重新延迟执行
                     readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
                 }
             }
         }

6. 项目实战(主要代码，项目以服务端作为心跳监控，也可以在客户端进行心跳监控)

   • 项目结构

     ├─src
     │  ├─main
     │  │  ├─java
     │  │  │  └─com
     │  │  │      └─hetangyuese
     │  │  │          └─netty
     │  │  │              ├─client
     │  │  │              │      MyChannelFutureListener.java
     │  │  │              │      MyClient05.java
     │  │  │              │      MyClientChannelHandler.java
     │  │  │              │      MyClientChannelInitializer.java
     │  │  │              │
     │  │  │              └─server
     │  │  │                  │  MyServer05.java
     │  │  │                  │  MyServerChannelInitializer.java
     │  │  │                  │  MyServerHandler.java
     │  │  │                  │
     │  │  │                  └─decoder

     ?

   • 在ChannelPipeline中注册IdleStateHandler

     public class MyServerChannelInitializer extends ChannelInitializer<SocketChannel> {
     
         @Override
         protected void initChannel(SocketChannel ch) throws Exception {
             ch.pipeline().
                     addLast(new StringEncoder(Charset.forName("GBK")))
                     .addLast(new StringDecoder(Charset.forName("GBK")))
                     .addLast(new LoggingHandler(LogLevel.INFO))
                    // 设置读取超时时间为5秒，写超时和总超时为0即不做监控
                     .addLast(new IdleStateHandler(5, 0, 0))
                     .addLast(new MyServerHandler());
         }
     }

   • 服务端处理handler(其中userEventTriggered为接收心跳任务触发的事件，这次做了计数三次触发读空闲超时则断开连接)

     public class MyServerHandler extends ChannelInboundHandlerAdapter {
     
         private AtomicInteger count = new AtomicInteger(1);
     
         /**
          *  心跳检测机制会进入
          * @param ctx
          * @param evt
          * @throws Exception
          */
         @Override
         public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
             System.out.println("心跳检测触发了事件, object: , time: " + evt + new Date().toLocaleString());
             super.userEventTriggered(ctx, evt);
             if (evt instanceof IdleStateEvent) {
                 IdleStateEvent e = (IdleStateEvent) evt;
                 // 客户端连接应该是请求 write
                 if (e.state() == IdleState.READER_IDLE) {
                     System.out.println("服务端监测到了读取超时");
                     count.incrementAndGet();
                     if (count.get() > 3) {
                         System.out.println("客户端还在？？ 已经3次检测没有访问了，我要断开了哦！！！");
                         ctx.channel().close();
                     }
                 } else if (e.state() == IdleState.WRITER_IDLE) {
                     // 如果一直有交互则会发送writer_idle
                     System.out.println("服务端收到了写入超时");
                 } else {
                     System.out.println("服务端收到了All_idle");
                 }
             } else {
                 super.userEventTriggered(ctx,evt);
             }
         }
     
         @Override
         public void channelActive(ChannelHandlerContext ctx) throws Exception {
             System.out.println("myServerHandler is active, time: " + new Date().toLocaleString());
             ctx.writeAndFlush("成功连接服务端, 当前时间：" + new Date().toLocaleString());
         }
     
         @Override
         public void channelInactive(ChannelHandlerContext ctx) throws Exception {
             System.out.println("服务端与客户端断开了连接, time: " + new Date().toLocaleString());
         }
     
         @Override
         public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
             System.out.println("myServerHandler 收到了客户端的信息 msg:" + msg + ", time: " + new Date().toLocaleString());
             ctx.writeAndFlush("您好，客户端，我是服务端");
         }
     
         @Override
         public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
             cause.printStackTrace();
             ctx.close();
         }
     }

   • 运行结果

     myServer is start time: 2019-11-8 14:47:16
     myServerHandler is active, time: 2019-11-8 14:47:18
     十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler channelRegistered
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] REGISTERED
     十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler channelActive
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] ACTIVE
     十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler write
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] WRITE: 成功连接服务端, 当前时间：2019-11-8 14:47:18
     十一月 08, 2019 2:47:18 下午 io.netty.handler.logging.LoggingHandler flush
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] FLUSH
     心跳检测触发了事件, object: , time:  14:47:23
     服务端监测到了读取超时
     心跳检测触发了事件, object: , time:  14:47:28
     服务端监测到了读取超时
     心跳检测触发了事件, object: , time:  14:47:33
     服务端监测到了读取超时
     客户端还在？？ 已经3次检测没有访问了，我要断开了哦！！！
     十一月 08, 2019 2:47:33 下午 io.netty.handler.logging.LoggingHandler close
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 - R:/127.0.0.1:62195] CLOSE
     十一月 08, 2019 2:47:33 下午 io.netty.handler.logging.LoggingHandler channelInactive
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 ! R:/127.0.0.1:62195] INACTIVE
     服务端与客户端断开了连接, time: 2019-11-8 14:47:33
     十一月 08, 2019 2:47:33 下午 io.netty.handler.logging.LoggingHandler channelUnregistered
     信息: [id: 0x8d924d06, L:/127.0.0.1:9001 ! R:/127.0.0.1:62195] UNREGISTERED

以上就是心跳监控的所有流程了，合理的利用Netty的心跳机制可以有效的剔除一些无用的连接释放些资源

Netty断线重连：在长连接中有时候出现断开的时候可以重新连接

出现断线重连的情况：

• 首次连接但是连接不上，通过ChannelFutureListener增加监控进行重连
• 由于网络原因等、自动断开等 通过在channelInactive中进行重连即可

1. ChannelFutureListener进行重连

   public class MyChannelFutureListener implements ChannelFutureListener {
   
       @Override
       public void operationComplete(ChannelFuture future) throws Exception {
           if (future.isSuccess()) {
               System.out.println("当前已连接");
               return;
           }
           System.out.println("启动连接客户端失败，开始重连");
           final EventLoop loop = future.channel().eventLoop();
           loop.schedule(new Runnable() {
               @Override
               public void run() {
                   try {
                      MyClient05.reConnection();
                      System.out.println("客户端重连成功");
                   } catch (Exception e){
                       e.printStackTrace();
                   }
               }
           }, 1L, TimeUnit.SECONDS);
       }
   }

   client启动 不知道为啥想要测试listener的效果时，connect的sync()不能带，是由于同步阻塞的原因？

   public void start() {
           EventLoopGroup group = new NioEventLoopGroup();
           try {
               bootstrap = getBootstrap();
   
               bootstrap.group(group)
                       .option(ChannelOption.AUTO_READ, true)
                       .option(ChannelOption.TCP_NODELAY, true)
                       .channel(NioSocketChannel.class)
                       .handler(new MyClientChannelInitializer());
   
               // ChannelFuture future = bootstrap.connect(new InetSocketAddress(ip, port)).sync();
                ChannelFuture future = bootstrap.connect(new InetSocketAddress(ip, port));      
               // 增加监听 
               future.addListener(new MyChannelFutureListener());
               future.channel().closeFuture().sync();
           } catch (Exception e) {
               e.printStackTrace();
           } finally {
               group.shutdownGracefully();
           }
       }

   直接启动客户端，可以看到listener输出：启动连接客户端失败，开始重连

2. channelInactive进行重连(我是直接new了一个线程去重连)

   @Override
   public void channelInactive(ChannelHandlerContext ctx) throws Exception {
       System.out.println("客户端断开了连接, time: " + new Date().toLocaleString());
       new Thread(new Runnable() {
           @Override
           public void run() {
               try {
                   new MyClient05("127.0.0.1", 9001).start();
                   System.out.println("客户端重新连接了服务端 time：" + new Date().toLocaleString());
               } catch (Exception e) {
                   e.printStackTrace();
               }
           }
       }).start();
   }

   测试方法：

   1.可以直接通过上面的心跳机制断开连接后，客户端的channelInactive检测到断开会自动执行重连

   测试结果：

   服务端：
   ------
   心跳检测触发了事件, object: , time:  16:43:12
   服务端监测到了读取超时
   心跳检测触发了事件, object: , time:  16:43:17
   服务端监测到了读取超时
   心跳检测触发了事件, object: , time:  16:43:22
   服务端监测到了读取超时
   客户端还在？？ 已经3次检测没有访问了，我要断开了哦！！！
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler close
   信息: [id: 0x6bfc0d90, L:/192.168.0.118:9001 - R:/192.168.0.118:51031] CLOSE
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelInactive
   信息: [id: 0x6bfc0d90, L:/192.168.0.118:9001 ! R:/192.168.0.118:51031] INACTIVE
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelUnregistered
   信息: [id: 0x6bfc0d90, L:/192.168.0.118:9001 ! R:/192.168.0.118:51031] UNREGISTERED
   服务端与客户端断开了连接, time: 2019-11-8 16:43:22
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelRegistered
   信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] REGISTERED
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler channelActive
   信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] ACTIVE
   myServerHandler is active, time: 2019-11-8 16:43:22
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler write
   信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] WRITE: 成功连接服务端, 当前时间：2019-11-8 16:43:22
   十一月 08, 2019 4:43:22 下午 io.netty.handler.logging.LoggingHandler flush
   信息: [id: 0xbcb2ec62, L:/127.0.0.1:9001 - R:/127.0.0.1:51061] FLUSH
   十一月 08, 2019 4:43:26 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
   ------------------------------------------------------------------------------------- 
   
   客户端：
   -------
   当前已连接
   客户端与服务端建立了连接 time: 2019-11-8 16:43:07
   客户端接收到了服务的响应的数据 msg: 成功连接服务端, 当前时间：2019-11-8 16:43:07, time: 2019-11-8 16:43:07
   客户端断开了连接, time: 2019-11-8 16:43:22
   当前已连接
   客户端与服务端建立了连接 time: 2019-11-8 16:43:22
   客户端接收到了服务的响应的数据 msg: 成功连接服务端, 当前时间：2019-11-8 16:43:22, time: 2019-11-8 16:43:22