心跳检测逻辑：服务端启动后，等待客户端连接，客户端连接之后，向服务端发送消息。如果客户端在线服务端必定会收到数据，如果客户端没在干活那么服务端接收不到客户端的消息。所以服务端检测一定时间内不活跃的客户端,将客户端连接关闭。Netty中自带了一个IdleStateHandler 可以用来实现心跳检测。
1、启动服务端，启动客户端
2、客户端死循环发送消息，并用sleep随机休眠时间，模拟空闲
3、服务端接收到消息，返回一条数据。客户端空闲时计数+1
4、服务端客户端继续通信
5、服务端检测客户端空闲态度，关闭连接。客户端发现连接被关闭了，就退出了。
心跳handler
package com.dmtest.netty_learn.heart_beat;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleStateEvent;
public class HeartBeatHandler extends SimpleChannelInboundHandler<String> {
int readIdleTimes = 0;
@Override
protected void channelRead0(ChannelHandlerContext ctx, String s) throws Exception {
System.out.println(" ====== > [server] message received : " + s);
if("I am alive".equals(s)){
ctx.channel().writeAndFlush("copy that");
}else {
System.out.println(" 其他信息处理 ... ");
}
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
IdleStateEvent event = (IdleStateEvent)evt;
String eventType = null;
switch (event.state()){
case READER_IDLE:
eventType = "读空闲";
readIdleTimes ++; // 读空闲的计数加1
break;
case WRITER_IDLE:
eventType = "写空闲";
// 不处理
break;
case ALL_IDLE:
eventType ="读写空闲";
// 不处理
break;
}
System.out.println("readIdleTimes="+readIdleTimes);
System.out.println(ctx.channel().remoteAddress() + "超时事件：" +eventType);
if(readIdleTimes > 3){
System.out.println(" [server]读空闲超过3次，关闭连接");
ctx.channel().writeAndFlush("you are out");
ctx.channel().close();
}
}
@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
}
}
服务端初始化
package com.dmtest.netty_learn.heart_beat;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.timeout.IdleStateHandler;
import java.util.concurrent.TimeUnit;
/**
* 2018/11/13.
*/
public class HeartBeatInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel channel) throws Exception {
ChannelPipeline pipeline = channel.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast(new IdleStateHandler(2,2,2, TimeUnit.SECONDS));
pipeline.addLast(new HeartBeatHandler());
}
}
pipeline.addLast(new IdleStateHandler(2,2,2, TimeUnit.SECONDS))，
三个空闲时间参数，以及时间参数的格式，我们例子设置的2,2,2。2秒没有读写，这个超时事件就会触发。
服务端代码
package com.dmtest.netty_learn.heart_beat;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
/**
* 2018/11/13.
*/
public class HeartBeatServer {
int port ;
public HeartBeatServer(int port){
this.port = port;
}
public void start(){
ServerBootstrap bootstrap = new ServerBootstrap();
EventLoopGroup boss = new NioEventLoopGroup();
EventLoopGroup worker = new NioEventLoopGroup();
try{
bootstrap.group(boss,worker)
.handler(new LoggingHandler(LogLevel.INFO))
.channel(NioServerSocketChannel.class)
.childHandler(new HeartBeatInitializer());
ChannelFuture future = bootstrap.bind(port).sync();
future.channel().closeFuture().sync();
}catch(Exception e){
e.printStackTrace();
}finally {
worker.shutdownGracefully();
boss.shutdownGracefully();
}
}
public static void main(String[] args) throws Exception {
HeartBeatServer server = new HeartBeatServer(8090);
server.start();
}
}
客户端代码
package com.dmtest.netty_learn.heart_beat;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
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;
/**
* 2018/11/13.
*/
public class HeartBeatClient {
int port;
Channel channel;
Random random ;
public HeartBeatClient(int port){
this.port = port;
random = new Random();
}
public static void main(String[] args) throws Exception{
HeartBeatClient client = new HeartBeatClient(8090);
client.start();
}
public void start() {
EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
try{
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
.handler(new HeartBeatClientInitializer());
connect(bootstrap,port);
String text = "I am alive";
while (channel.isActive()){
sendMsg(text);
}
}catch(Exception e){
// do something
}finally {
eventLoopGroup.shutdownGracefully();
}
}
public void connect(Bootstrap bootstrap,int port) throws Exception{
channel = bootstrap.connect("localhost",8090).sync().channel();
}
public void sendMsg(String text) throws Exception{
channel.writeAndFlush(text);
int num = random.nextInt(10);
System.out.println(" client sent msg and sleep "+num);
Thread.sleep(num * 1000);
}
static class HeartBeatClientInitializer extends ChannelInitializer<Channel> {
@Override
protected void initChannel(Channel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast("decoder", new StringDecoder());
pipeline.addLast("encoder", new StringEncoder());
pipeline.addLast(new HeartBeatClientHandler());
}
}
static class HeartBeatClientHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
System.out.println(" client received :" +msg);
if(msg!= null && msg.equals("you are out")) {
System.out.println(" server closed connection , so client will close too");
ctx.channel().closeFuture();
}
}
}
}
客户端
服务端
参考
Netty实现心跳机制
netty心跳代码地址
netty中的心跳检测机制
springboot整合 netty做心跳检测1
