心跳检测
客户端和服务端的连接属于socket连接，也属于长连接，往往会存在客户端在连接了服务端之后就没有任何操作了，但还是占用了一个连接；当越来越多类似的客户端出现就会浪费很多连接，netty中可以通过心跳检测来找出一定程度（自定义规则判断哪些连接是无效链接）的无效链接并断开连接，保存真正活跃的连接。
我理解的心跳检测应该是客户端/服务端定时发送一个数据包给服务端/客户端，检测对方是否有响应；
如果是存活的连接，在一定的时间内应该会收到响应回来的数据包；
如果在一定时间内还是收不到接收方的响应的话，就可以当做是挂机，可以断开此连接；
如果检测到了掉线之后还可以进行重连；
- TCP自带心跳检测，协议层采用Keeplive机制默认2小时频率触发一次检测，但是它存在缺陷:检测不出网线拔出、防火墙、使用起来不灵活、依赖操作系统等
- Netty可以通过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的心跳机制可以有效的剔除一些无用的连接释放些资源
- 首次连接但是连接不上，通过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