响应式网站设计软件,广告公司企业简介,wordpress上帝模式,wordpress 慢 google优质博文#xff1a;IT-BLOG-CN
一、心跳机制
Netty支持心跳机制#xff0c;可以检测远程服务端是否存活或者活跃。心跳是在TCP长连接中#xff0c;客户端和服务端定时向对方发送数据包通知对方自己还在线#xff0c;保证连接的有效性的一种机制。在服务器和客户端之间一…优质博文IT-BLOG-CN
一、心跳机制
Netty支持心跳机制可以检测远程服务端是否存活或者活跃。心跳是在TCP长连接中客户端和服务端定时向对方发送数据包通知对方自己还在线保证连接的有效性的一种机制。在服务器和客户端之间一定时间内没有数据交互时, 即处于idle状态时, 客户端或服务器会发送一个特殊的数据包给对方, 当接收方收到这个数据报文后, 也立即发送一个特殊的数据报文, 回应发送方, 即一个PING-PONG交互。当某一端收到心跳消息后, 就知道了对方仍然在线, 这就确保TCP连接的有效性。
Netty提供了IdleStateHandler可以对三种类型心跳进行检测是用来监测连接的空闲情况。然后我们就可以根据心跳情况来实现具体的处理逻辑比如说断开连接、重新连接等等。同时还提供了ReadTimeoutHandlerWriteTimeoutHandler检测连接的有效性。
名称作用IdleStateHandler当连接的空闲时间读或者写太长时将会触发一个IdleStateEvent事件。然后你可以通过你的ChannelInboundHandler中重写userEventTrigged方法来处理该事件。ReadTimeoutHandler如果在指定的事件没有发生读事件就会抛出这个异常并自动关闭这个连接。你可以在exceptionCaught方法中处理这个异常。WriteTimeoutHandler当一个写操作不能在一定的时间内完成时抛出此异常并关闭连接。你同样可以在exceptionCaught方法中处理这个异常。
二、IdleStateHandler 简介 IdleStateHandler也是一个ChannelHandler也需要被载入到ChannelPipeline中加入我们在服务器端的ChannelInitializer中。我们在channel链中加入了IdleSateHandler第一个参数是5单位是秒那么这样做的意思就是在服务器端会每隔5秒来检查一下channelRead方法被调用的情况如果在5秒内该链上的channelRead方法都没有被触发就会调用userEventTriggered方法
//创建服务类
ServerBootstrap serverBootstrap new ServerBootstrap();//boss和worker
NioEventLoopGroup boss new NioEventLoopGroup();
NioEventLoopGroup worker new NioEventLoopGroup();try {//设置线程池serverBootstrap.group(boss,worker);//设置socket工厂,Channel 是对 Java 底层 Socket 连接的抽象serverBootstrap.channel(NioServerSocketChannel.class);//设置管道工厂serverBootstrap.childHandler(new ChannelInitializerChannel() {Overrideprotected void initChannel(Channel ch) throws Exception {//设置后台转换器二进制转换字符串ch.pipeline().addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));ch.pipeline().addLast(new StringDecoder());ch.pipeline().addLast(new StringEncoder());ch.pipeline().addLast(new ServerSocketHandler());}});并且还是个ChannelInboundHandler是用来处理入站事件的。看下它的构造器
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事件
三、IdleStateHandler 源码
【1】handlerAdded和handlerRemovedIdleStateHandler是在创建IdleStateHandler实例并添加到ChannelPipeline时添加定时任务来进行定时检测的具体在initialize(ctx)方法实现同时在从ChannelPipeline移除或Channel关闭时移除这个定时检测具体在destroy()实现。IdleStateHandler的channelActive()方法在socket通道建立时被触发。
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {if (ctx.channel().isActive() ctx.channel().isRegistered()) {this.initialize(ctx);}}public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {this.destroy();
}public void channelInactive(ChannelHandlerContext ctx) throws Exception {this.destroy();super.channelInactive(ctx);
}【2】initialize根据配置的readerIdleTimeWriteIdleTIme等超时事件参数往任务队列taskQueue中添加定时任务task。我先先查看ReaderIdleTimeoutTask的作用。
private void initialize(ChannelHandlerContext ctx) {switch(this.state) {case 1:case 2:return;default:this.state 1;this.initOutputChanged(ctx);this.lastReadTime this.lastWriteTime this.ticksInNanos();if (this.readerIdleTimeNanos 0L) {// 这里的 schedule 方法会调用 eventLoop 的 schedule 方法将定时任务添加进队列中this.readerIdleTimeout this.schedule(ctx, new IdleStateHandler.ReaderIdleTimeoutTask(ctx), this.readerIdleTimeNanos, TimeUnit.NANOSECONDS);}if (this.writerIdleTimeNanos 0L) {this.writerIdleTimeout this.schedule(ctx, new IdleStateHandler.WriterIdleTimeoutTask(ctx), this.writerIdleTimeNanos, TimeUnit.NANOSECONDS);}if (this.allIdleTimeNanos 0L) {this.allIdleTimeout this.schedule(ctx, new IdleStateHandler.AllIdleTimeoutTask(ctx), this.allIdleTimeNanos, TimeUnit.NANOSECONDS);}}
}定时任务添加到对应线程EventLoopExecutor对应的任务队列taskQueue中在对应线程的run()方法中循环执行。只要给定的参数大于0就创建一个定时任务每个事件都创建。同时将state状态设置为1防止重复初始化。调用initOutputChanged方法初始化监控出站数据属性
private void initOutputChanged(ChannelHandlerContext ctx) {if (this.observeOutput) {Channel channel ctx.channel();Unsafe unsafe channel.unsafe();ChannelOutboundBuffer buf unsafe.outboundBuffer();// 记录了出站缓冲区相关的数据buf 对象的 hash 码和 buf 的剩余缓冲字节数if (buf ! null) {this.lastMessageHashCode System.identityHashCode(buf.current());this.lastPendingWriteBytes buf.totalPendingWriteBytes();this.lastFlushProgress buf.currentProgress();}}
}这边会触发一个ReaderIdleTimeoutTasknextDelay的初始化值为超时秒数readerIdleTimeNanos。如果检测的时候没有正在读且计算多久没读了nextDelay - 当前时间 - 上次读取时间假如这个结果是6s说明最后一次调用channelRead已经是6s之前的事情了你设置的是5s那么nextDelay则为-1说明超时了。则创建IdleStateEvent事件IdleState枚举值为READER_IDLE然后调用channelIdle方法分发给下一个ChannelInboundHandler通常由用户自定义一个ChannelInboundHandler来捕获并处理。
private final class ReaderIdleTimeoutTask extends IdleStateHandler.AbstractIdleTask {ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {super(ctx);}protected void run(ChannelHandlerContext ctx) {long nextDelay IdleStateHandler.this.readerIdleTimeNanos;if (!IdleStateHandler.this.reading) {nextDelay - IdleStateHandler.this.ticksInNanos() - IdleStateHandler.this.lastReadTime;}if (nextDelay 0L) {IdleStateHandler.this.readerIdleTimeout IdleStateHandler.this.schedule(ctx, this, IdleStateHandler.this.readerIdleTimeNanos, TimeUnit.NANOSECONDS);boolean first IdleStateHandler.this.firstReaderIdleEvent;IdleStateHandler.this.firstReaderIdleEvent false;try {IdleStateEvent event IdleStateHandler.this.newIdleStateEvent(IdleState.READER_IDLE, first);IdleStateHandler.this.channelIdle(ctx, event);} catch (Throwable var6) {ctx.fireExceptionCaught(var6);}} else {IdleStateHandler.this.readerIdleTimeout IdleStateHandler.this.schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);}}
}firstxxxxIdleEvent作用假设当你的客户端应用每次接收数据是30秒而你的写空闲时间是25秒那么当你数据还没有写出的时候写空闲时间触发了。实际上是不合乎逻辑的。因为你的应用根本不空闲。
Netty的解决方案是记录最后一次输出消息的相关信息并使用一个值firstXXXXIdleEvent表示是否再次活动过每次读写活动都会将对应的first值更新为true如果是false说明这段时间没有发生过读写事件。同时如果第一次记录出站的相关数据和第二次得到的出站相关数据不同则说明数据在缓慢的出站就不用触发空闲事件。
总的来说这个字段就是用来对付 “客户端接收数据奇慢无比慢到比空闲时间还多” 的极端情况。所以Netty默认是关闭这个字段的。
总的来说每次读取操作都会记录一个时间定时任务时间到了会计算当前时间和最后一次读的时间的间隔如果间隔超过了设置的时间就触发·UserEventTriggered方法
protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {ctx.fireUserEventTriggered(evt);
}【3】写任务的run方法逻辑基本和读任务的逻辑一样唯一不同的就是有一个针对 出站较慢数据的判断。
if (hasOutputChanged(ctx, first)) {return;
}如果这个方法返回true就不执行触发事件操作了即使时间到了。看看该方法实现
private boolean hasOutputChanged(ChannelHandlerContext ctx, boolean first) {if (observeOutput) {// 如果最后一次写的时间和上一次记录的时间不一样说明写操作进行过了则更新此值if (lastChangeCheckTimeStamp ! lastWriteTime) {lastChangeCheckTimeStamp lastWriteTime;// 但如果在这个方法的调用间隙修改的就仍然不触发事件if (!first) { // #firstWriterIdleEvent or #firstAllIdleEventreturn true;}}Channel channel ctx.channel();Unsafe unsafe channel.unsafe();ChannelOutboundBuffer buf unsafe.outboundBuffer();// 如果出站区有数据if (buf ! null) {// 拿到出站缓冲区的 对象 hashcodeint messageHashCode System.identityHashCode(buf.current());// 拿到这个 缓冲区的 所有字节long pendingWriteBytes buf.totalPendingWriteBytes();// 如果和之前的不相等或者字节数不同说明输出有变化将 最后一个缓冲区引用 和 “剩余字节数” 刷新if (messageHashCode ! lastMessageHashCode || pendingWriteBytes ! lastPendingWriteBytes) {lastMessageHashCode messageHashCode;lastPendingWriteBytes pendingWriteBytes;// 如果写操作没有进行过则任务写的慢不触发空闲事件if (!first) {return true;}}}}return false;
}如果用户没有设置需要观察出站情况。就返回false继续执行事件。如果设置了观察出站的情况且最后一次写的时间和上一次记录的时间不一样说明写操作刚刚做过了则更新此值但仍然需要判断这个first的值如果这个值还是false说明在这个写事件是在两个方法调用间隙完成的或者是第一次访问这个方法就仍然不触发事件。如果不满足上面的条件就取出缓冲区对象如果缓冲区没对象了说明没有发生写的很慢的事件就触发空闲事件。反之记录当前缓冲区对象的hashcode和剩余字节数再和之前的比较如果任意一个不相等说明数据在变化或者说数据在慢慢的写出去。那么就更新这两个值留在下一次判断。继续判断first如果是fasle说明这是第二次调用就不用触发空闲事件了。 【4】所有事件的run方法这个类叫做AllIdleTimeoutTask表示这个监控着所有的事件。当读写事件发生时都会记录。代码逻辑和写事件的的基本一致除了这里
long nextDelay allIdleTimeNanos;
if (!reading) {// 当前时间减去 最后一次写或读 的时间 若大于0说明超时了nextDelay - ticksInNanos() - Math.max(lastReadTime, lastWriteTime);
}这里的时间计算是取读写事件中的最大值来的。然后像写事件一样判断是否发生了写的慢的情况。最后调用ctx.fireUserEventTriggered(evt)方法。
通常这个使用的是最多的。构造方法一般是
pipeline.addLast(new IdleStateHandler(0, 0, 30, TimeUnit.SECONDS));读写都是0表示禁用30表示30秒内没有任务读写事件发生就触发事件。注意当不是0的时候这三个任务会重叠。
四、总结
【1】IdleStateHandler心跳检测主要是通过向线程任务队列中添加定时任务判断channelRead()方法或write()方法是否调用空闲超时如果超时则触发超时事件执行自定义userEventTrigger()方法 【2】Netty通过IdleStateHandler实现最常见的心跳机制不是一种双向心跳的PING-PONG模式而是客户端发送心跳数据包服务端接收心跳但不回复因为如果服务端同时有上千个连接心跳的回复需要消耗大量网络资源如果服务端一段时间内没有收到客户端的心跳数据包则认为客户端已经下线将通道关闭避免资源的浪费在这种心跳模式下服务端可以感知客户端的存活情况无论是宕机的正常下线还是网络问题的非正常下线服务端都能感知到而客户端不能感知到服务端的非正常下线 【3】要想实现客户端感知服务端的存活情况需要进行双向的心跳Netty中的channelInactive()方法是通过Socket连接关闭时挥手数据包触发的因此可以通过channelInactive()方法感知正常的下线情况但是因为网络异常等非正常下线则无法感知
