devbox
羊村懒王
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

Netty4实现心跳检测及IdleStateHandler源码分析_idlestatehandler 心跳检测 redis是否连接正常代码怎么写

作者：羊村懒王 | 2024-03-15 15:10:54

赞

踩

idlestatehandler 心跳检测 redis是否连接正常代码怎么写

1.什么是心跳检测？

    判断对方是否正常运行，一般采用定时发送简单的通讯包，如果在指定时间内未接收到对方响应，则判定对方已经宕掉。用于检测TCP的异常断开。

    心跳包一般就是客户端发送给服务端的简单消息，如果服务端几分钟内没有收到客户端消息，则视为客户端已经断开，这个时候就主动关闭客户端的通道。

 

2.使用Netty实现服务端心跳检测

    下面我们编写服务端代码，服务端实现以下功能：如果在N长时间内没有接受到客户端连接，则发送一段信息给客户端，并关闭其通道

 

    * 我们创建服务端心跳检测的Handler，命名为HeartBeatHandler，并重写userEventTriggered方法

  1. /**
  2.  * 服务端心跳检测
  3.  */
  4. public class HeartBeatHandler extends ChannelInboundHandlerAdapter {
  5.  
  6. 	@Override
  7. 	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
  8.  
  9. 		if(evt instanceof IdleStateEvent){
  10. 			IdleStateEvent ise = (IdleStateEvent)evt;
  11. 			
  12. 			// 服务端读空闲
  13. 			if(ise.state().equals(IdleState.READER_IDLE)){
  14. 				ctx.writeAndFlush("client reader idle, channel will close");
  15. 				ctx.channel().close();
  16. 				
  17. 			// 服务端写空闲
  18. 			}else if (ise.state().equals(IdleState.WRITER_IDLE)){
  19. 				ctx.write("pong message");
  20. 				
  21. 			// 服务端读写空闲
  22. 			}else if (ise.state().equals(IdleState.ALL_IDLE)){
  23. 				ctx.channel().close();
  24. 			}else{
  25. 				// DO NOTHING
  26. 			}
  27. 		}
  28. 	}
  29. }

    * 服务端代码，如下

  1. public class Server {
  2.  
  3. 	public static void main(String[] args) {
  4. 		//服务类
  5. 		ServerBootstrap bootstrap = new ServerBootstrap();
  6. 		
  7. 		//boss和worker
  8. 		EventLoopGroup boss = new NioEventLoopGroup();
  9. 		EventLoopGroup worker = new NioEventLoopGroup();
  10. 		
  11. 		try {
  12. 			//设置线程池
  13. 			bootstrap.group(boss, worker);
  14. 			
  15. 			//设置socket工厂
  16. 			bootstrap.channel(NioServerSocketChannel.class);
  17. 			
  18. 			//设置管道工厂
  19. 			bootstrap.childHandler(new ChannelInitializer<Channel>() {
  20.  
  21.                
  22. 				@Override
  23. 				protected void initChannel(Channel ch) throws Exception {
  24. 					ch.pipeline().addLast(new StringEncoder());
  25.                      // 主要在这里，先创建一个IdleStateHandler，10秒内没有读写则判定为空闲
  26. 					ch.pipeline().addLast(new IdleStateHandler(10, 10, 10, TimeUnit.SECONDS));
  27.                     // 然后执行心跳检测
  28. 					ch.pipeline().addLast(new HeartBeatHandler());
  29. 				}
  30. 			});
  31. 			
  32. 			//设置参数，TCP参数
  33. 			bootstrap.option(ChannelOption.SO_BACKLOG, 2048);//serverSocketchannel的设置，链接缓冲池的大小
  34. 			bootstrap.childOption(ChannelOption.SO_KEEPALIVE, true);//socketchannel的设置,维持链接的活跃，清除死链接
  35. 			bootstrap.childOption(ChannelOption.TCP_NODELAY, true);//socketchannel的设置,关闭延迟发送
  36. 			
  37. 			//绑定端口
  38. 			ChannelFuture future = bootstrap.bind(8088).sync();
  39. 			
  40. 			System.out.println("server start...");
  41. 			//等待服务端关闭
  42. 			future.channel().closeFuture().sync();
  43. 			
  44. 		} catch (Exception e) {
  45. 			e.printStackTrace();
  46. 		} finally{
  47. 			//释放资源
  48. 			boss.shutdownGracefully();
  49. 			worker.shutdownGracefully();
  50. 		}
  51. 	}
  52. }

    * 启动服务端并测试

    启动服务端之后，我们来开启一个客户端，还是使用telnet的方式

    连接到服务端之后，如果我们十秒钟不发送数据的话，则会被强制关闭连接，如下图所示

 

3.关于IdleStateHandler源码分析

 

    1）IdleStateHandler结构分析

  1. public class IdleStateHandler extends ChannelDuplexHandler {
  2.     
  3.     // 对象初始化的时候初始化好这些数据
  4.     private final long readerIdleTimeNanos;
  5.     private final long writerIdleTimeNanos;
  6.     private final long allIdleTimeNanos;
  7.  
  8.     // 通过构造函数可知
  9.     public IdleStateHandler(
  10.             long readerIdleTime, long writerIdleTime, long allIdleTime,
  11.             TimeUnit unit) {
  12.         if (unit == null) {
  13.             throw new NullPointerException("unit");
  14.         }
  15.  
  16.         if (readerIdleTime <= 0) {
  17.             readerIdleTimeNanos = 0;
  18.         } else {
  19.             readerIdleTimeNanos = Math.max(unit.toNanos(readerIdleTime), MIN_TIMEOUT_NANOS);
  20.         }
  21.         if (writerIdleTime <= 0) {
  22.             writerIdleTimeNanos = 0;
  23.         } else {
  24.             writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
  25.         }
  26.         if (allIdleTime <= 0) {
  27.             allIdleTimeNanos = 0;
  28.         } else {
  29.             allIdleTimeNanos = Math.max(unit.toNanos(allIdleTime), MIN_TIMEOUT_NANOS);
  30.         }
  31.     }
  32.     
  33.     // 在channelRegistered/channelActive的时候调用初始化方法initialize方法
  34.     @Override
  35.     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
  36.         // Initialize early if channel is active already.
  37.         if (ctx.channel().isActive()) {
  38.             initialize(ctx);
  39.         }
  40.         super.channelRegistered(ctx);
  41.     }
  42.  
  43.     @Override
  44.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  45.         // This method will be invoked only if this handler was added
  46.         // before channelActive() event is fired.  If a user adds this handler
  47.         // after the channelActive() event, initialize() will be called by beforeAdd().
  48.         initialize(ctx);
  49.         super.channelActive(ctx);
  50.     }
  51.     
  52.     // 监听到客户端写数据的时候，更新lastReadTime
  53.     @Override
  54.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  55.         lastReadTime = System.nanoTime();
  56.         firstReaderIdleEvent = firstAllIdleEvent = true;
  57.         ctx.fireChannelRead(msg);
  58.     }

    总结：通过以上对IdleStateHandler的分析，可知：

    * 在构造IdleStateHandler的时候，便使用创建参数确定读、写、读写的空闲时间，并转换为millsecond

    * 当客户端连接到服务端，并且其channel处于活动状态，则调用initialize初始化方法

    * 当服务端监听到客户端发送数据时，即channelRead方法，则更新lastReadTime参数，即最近一次读时间

 

    可知，最重要的逻辑应该再initialize方法中，下面我们就来分析一下该方法

 

    2）initialize()

  1. private void initialize(ChannelHandlerContext ctx) {
  2.     // Avoid the case where destroy() is called before scheduling timeouts.
  3.     // See: https://github.com/netty/netty/issues/143
  4.     
  5.     // private volatile int state; // 0 - none, 1 - initialized, 2 - destroyed
  6.     switch (state) {
  7.         case 1:
  8.         case 2:
  9.             return;
  10.     }
  11.  
  12.     state = 1;
  13.  
  14.     // 1.获取线程池
  15.     EventExecutor loop = ctx.executor();
  16.  
  17.     // 2.初始化最后一次读时间和最后一次写时间为当前时间
  18.     lastReadTime = lastWriteTime = System.nanoTime();
  19.     
  20.     // 3.根据用户设置的读空闲时间启动一个定时任务，读空闲时间为频率执行
  21.     if (readerIdleTimeNanos > 0) {
  22.         readerIdleTimeout = loop.schedule(
  23.             new ReaderIdleTimeoutTask(ctx),
  24.             readerIdleTimeNanos, TimeUnit.NANOSECONDS);
  25.     }
  26.     
  27.     // 4.根据用户设置的写空闲时间启动一个定时任务，写空闲时间为频率执行
  28.     if (writerIdleTimeNanos > 0) {
  29.         writerIdleTimeout = loop.schedule(
  30.             new WriterIdleTimeoutTask(ctx),
  31.             writerIdleTimeNanos, TimeUnit.NANOSECONDS);
  32.     }
  33.     
  34.     // 5.根据用户设置的读写空闲时间启动一个定时任务，读写空闲时间为频率执行
  35.     if (allIdleTimeNanos > 0) {
  36.         allIdleTimeout = loop.schedule(
  37.             new AllIdleTimeoutTask(ctx),
  38.             allIdleTimeNanos, TimeUnit.NANOSECONDS);
  39.     }
  40. }

    总结：由上可知，实现心跳检测的关键就在这个schedule上，根据用户设置的空闲时间来确定定时任务的频率。

    那么，定时任务是如何做到检测的呢？我们继续来看

 

    3）ReaderIdleTimeoutTask

    源码如下：

  1. private final class ReaderIdleTimeoutTask implements Runnable {
  2.  
  3.     private final ChannelHandlerContext ctx;
  4.  
  5.     ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
  6.         this.ctx = ctx;
  7.     }
  8.  
  9.     @Override
  10.     public void run() {
  11.         if (!ctx.channel().isOpen()) {
  12.             return;
  13.         }
  14.  
  15.         // 1.获取当前时间和最后一次读时间
  16.         long currentTime = System.nanoTime();
  17.         long lastReadTime = IdleStateHandler.this.lastReadTime;
  18.         
  19.         // nextDelay即为比较用户设置的读空闲时间和 当前时间-最后一次时间
  20.         long nextDelay = readerIdleTimeNanos - (currentTime - lastReadTime);
  21.         
  22.         // 2.如果nextDelay<=0，则说明客户端已经在超过读空闲时间内没有写入数据了
  23.         if (nextDelay <= 0) {
  24.             // 2.1 重新定义readerIdleTimeout schedule，与initialize方法设置的相同，继续执行定时任务
  25.             readerIdleTimeout =
  26.                 ctx.executor().schedule(this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
  27.             try {
  28.                 // 2.2 event = new IdleStateEvent(IdleState.READER_IDLE, true)，将event设置为读空闲
  29.                 IdleStateEvent event;
  30.                 if (firstReaderIdleEvent) {
  31.                     firstReaderIdleEvent = false;
  32.                     event = IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT;
  33.                 } else {
  34.                     event = IdleStateEvent.READER_IDLE_STATE_EVENT;
  35.                 }
  36.                 // 2.3 channelIdle的主要工作就是将evt传输给下一个Handler
  37.                 channelIdle(ctx, event);
  38.             } catch (Throwable t) {
  39.                 ctx.fireExceptionCaught(t);
  40.             }
  41.             
  42.         // 3.如果nextDelay>0，则说明客户端在规定时间内已经写入数据了，
  43.         } else {
  44.             // 3.1 重新定义readerIdleTimeout schedule，以nextDelay为执行频率
  45.             readerIdleTimeout = ctx.executor().schedule(this, nextDelay, TimeUnit.NANOSECONDS);
  46.         }
  47.     }
  48. }

    由以上源码分析可知，ReaderIdleTimeoutTask的主要功能就是：

    * 根据当前时间和用户最后一次读时间来确实用户是否写入超时

    * 如果已经写超时，则重新定义schedule任务，还是以readerIdleTimeNanos为频率来执行。并设置evt为READ_IDLE，并传递给下一个Handler

    * 如果没有写超时，则继续执行schedule任务，此时以nextDelay（readerIdleTimeNanos - (currentTime - lastReadTime)）为频率来执行

    

    4）后续处理

    READ_IDLE事件传递给下一个Handler，在本例中即为传递给HeartBeatHandler，其重新定义了userEventTriggered方法，此时正好接收到READ_IDLE事件，做相应处理即可

 

 

参考：Netty in Action

 

 

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/羊村懒王/article/detail/242167
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号