netty实现心跳检测
检测逻辑：
1） 服务端启动，客户端建立连接，连接的目的是互相发送消息。
2） 如果客户端在工作，服务端一定能收到数据，如果客户端空闲，服务端会出现资源浪费。
3） 服务端需要一种检测机制，验证客户端的活跃状态，不活跃则关闭。
需求设计：
1） 客户端向服务端发送 “I am alive” , sleep一个随机时间，模拟空闲状态
2） 服务端收到消息后，返回“over”， 客户端有空闲，记录空闲次数
3） 设定阈值，达到阈值时主动关闭连接
服务端编写
public class HreatBeatServer {
public static void main(String[] args) {
//创建两个Reactor 构建主从 Reactor 模型
//用于处理 连接和读写事件 ， 无限循环组（线程池）
//管理 channel 监听事件
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
// 我们需要一个服务端引导程序来开启服务端。
ServerBootstrap serverBootstrap = new ServerBootstrap();
//将主从 Reactor 入参，设置当前参数
//这个方法返回的事对象本身，我们可以点出其他方法， 这种返回类型为对象自身 提供了 链式编程的方式
serverBootstrap.group(bossGroup, workerGroup)
//我们需要设置 channel 的 类型
//对应的是 netty NIO BIO
//NioServerSocketChannel <== ServerSocketChannel <== ServerSocket
.channel(NioServerSocketChannel.class)
//设置当前通道的处理器，使用Netty提供的日志打印处理器
.handler(new LoggingHandler(LogLevel.INFO))
//定义客户端连接处理的使用
//此方法需要设置参数 ChannelInitializer 通道初始化器
//初始化 要处理客户端 通道， 所以泛型设置为 SocketChannel
//此类 为抽象类 需要实现其抽象方法 initchannel （alt+enter）快捷键
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//通过channel 获取管道 pipeline
// 通道代表我们连接的角色， 管道代表处理业务得逻辑管理
// 管道相当于 链表，可以将不同的处理器连接起来，管理处理器的顺序
// 使用时 常常使用的事尾插法， addList 将加入到尾部
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
/*
* 使用心跳检测处理器
* 读空闲 写空闲 读写空闲 的超时时间
* 最后一个参数是 时间的单位
* IdleStateHandler发现有空闲的时候 会触发 IdleStateEvent时间
* 他会把事件推送给下一个 handler的指定方法 userEventTriggered 去处理
* */
pipeline.addLast(new IdleStateHandler(5, 10, 20, TimeUnit.SECONDS));
socketChannel.pipeline().addLast(new HreatBeatServerHandler());
}
});
System.out.println("服务端初始化完成");
// 设置并启动端口号，但需要使用sync 异步启动
try {
ChannelFuture future = serverBootstrap.bind(2020).sync();
// 将关闭通道的方式也设置为异步的
// 阻塞Finally中的代码执行
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
// 优雅方式都关闭
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
IdleStateHandler , 是netty提供的处理器
1）超过多长时间没有读 readerIdleTime
2) 超过多长时间没有写 writerIdleTime
3) 超过多长时间没有读和写 allIdleTime
底层实现检测的是 IdleStateEvent事件，通过管道传递给下一个handler处理，处理方法是userEventTriggered。
处理器编写
public class HreatBeatServerHandler extends SimpleChannelInboundHandler<String> {
private int times;
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
if ("I am alive".equals(msg)) {
ctx.writeAndFlush(Unpooled.copiedBuffer("over", CharsetUtil.UTF_8));
}
}
//处理心跳检测事件的方法
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
IdleStateEvent event = (IdleStateEvent) evt;
String eventDesc = null;
switch (event.state()) {
case READER_IDLE:
eventDesc = "读空闲";
break;
case WRITER_IDLE:
eventDesc = "写空闲";
break;
case ALL_IDLE:
eventDesc = "读写空闲";
break;
}
System.out.println(ctx.channel().remoteAddress() + "发生超时事件--" + eventDesc);
times++;
if (times > 3) {
System.out.println("空闲次数超过三次 关闭连接");
ctx.writeAndFlush("you are out");
ctx.channel().close();
}
//super.userEventTriggered(ctx, evt);
}
}
其中IdleStateEvent事件，分为READER_IDLE、WRITER_IDLE、ALL_IDLE三大类
客户端编写
客户端不断循环给服务端发消息确认存活的期间 线程睡眠 模拟失去心跳场景
package com.hyc.netty.Hreatbeat;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import java.util.Random;
public class HreatbeatClient {
public static void main(String[] args) {
//客户端只需要一个事件循环组
EventLoopGroup group = new NioEventLoopGroup();
// 客户端启动的对象
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
ChannelPipeline pipeline = socketChannel.pipeline();
pipeline.addLast(new StringDecoder());
pipeline.addLast(new StringEncoder());
socketChannel.pipeline().addLast(new HreatbeatClientHandler());
}
});
System.out.println("客户端初始化完成");
try {
ChannelFuture future = bootstrap.connect("127.0.0.1", 2020).sync();
String data = "I am alive";
while (future.channel().isActive()) {
//模拟空闲状态
int num = new Random().nextInt(10);
Thread.sleep(num * 1000);
future.channel().writeAndFlush(data);
}
//future.channel().closeFuture().sync();
} catch (Exception e) {
e.printStackTrace();
} finally {
group.shutdownGracefully();
}
}
static class HreatbeatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, String s) throws Exception {
System.out.println("server data:" + s);
if ("you are out".equals(s)) {
System.out.println("关闭");
channelHandlerContext.channel().close();
}
}
}
}
客户端随机线程睡眠 一旦接受到 服务端返回的you are out代表空闲次数超过了 3次 则关闭客户端连接