Netty学习篇④-心跳机制及断线重连
心跳检测
-
前言
客户端和服务端的连接属于socket连接，也属于长连接，往往会存在客户端在连接了服务端之后就没有任何操作了，但还是占用了一个连接；当越来越多类似的客户端出现就会浪费很多连接，netty中可以通过心跳检测来找出一定程度（自定义规则判断哪些连接是无效链接）的无效链接并断开连接，保存真正活跃的连接。
-
什么叫心跳检测
我理解的心跳检测应该是客户端/服务端定时发送一个数据包给服务端/客户端，检测对方是否有响应； 如果是存活的连接，在一定的时间内应该会收到响应回来的数据包； 如果在一定时间内还是收不到接收方的响应的话，就可以当做是挂机，可以断开此连接； 如果检测到了掉线之后还可以进行重连；
-
心跳检测的实现
- TCP自带心跳检测，协议层采用Keeplive机制默认2小时频率触发一次检测，但是它存在缺陷:检测不出网线拔出、防火墙、使用起来不灵活、依赖操作系统等
- Netty可以通过IdleStateHandler来实现心跳检测，使用起来也非常方便清晰
-
IdleStateHandler原理
idleStateHandler在通道注册之后会开启一个定时任务，定时去检测通道中后续是否还有进行数据传输，如果在规定的时间内没有进行数据传输则会触发对应的超时事件，使用者可以根据对应的事件自定义规则来判别当前连接是否是活跃，是否需要关闭连接等来进行操作。 一般idleStateHandler触发的事件IdleStateEvent会在心跳handler中的userEventTriggered方法中捕获到对应的超时事件。
IdleStateHandler的继承关系：通过ChannelDuplexHandler类继承ChannelInboundHandler和实现ChannelOutboundHandler来实现对入站和出站的重写和监控
-
源码分析
-
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); } } }
-
-
项目实战(主要代码，项目以服务端作为心跳监控，也可以在客户端进行心跳监控)
-
项目结构
├─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: io.netty.handler.timeout.IdleStateEvent@a3a0212019-11-8 14:47:23 服务端监测到了读取超时 心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@18692702019-11-8 14:47:28 服务端监测到了读取超时 心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@18692702019-11-8 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中进行重连即可
-
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输出：启动连接客户端失败，开始重连
-
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: io.netty.handler.timeout.IdleStateEvent@ab81552019-11-8 16:43:12 服务端监测到了读取超时 心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@15192622019-11-8 16:43:17 服务端监测到了读取超时 心跳检测触发了事件, object: , time: io.netty.handler.timeout.IdleStateEvent@15192622019-11-8 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
正文到此结束
热门推荐
相关文章
近期评论
-
文章和留言都翻到11页了 没有OOM
-
-
朋友，翻页到11页，及以后，会出现OOM，无法访问
-
-
搞个gitee的项目
-
-
版本号是多少，你可以下载哪个代码仓库，jdk选1.8 直接跑就行
-
-
-
Loading...
![[HBLOG]公众号](https://www.liuhaihua.cn/img/qrcode_gzh.jpg)
