Netty中的心跳检测机制详解

更新时间：2023年12月11日 11:06:41 作者：warybee

这篇文章主要介绍了Netty中的心跳检测机制详解,Netty 是基于 TCP 协议开发的,在四层协议 TCP 协议的实现中也提供了 keepalive 报文用来探测对端是否可用,TCP 层将在定时时间到后发送相应的 KeepAlive 探针以确定连接可用性,需要的朋友可以参考下

Netty心跳检测机制

1 心跳检测使用场景

长连接的应用场景非常的广泛，比如监控系统，IM系统，即时报价系统，推送服务等等。像这些场景都是比较注重实时性，如果每次发送数据都要进行一次DNS解析，建立连接的过程肯定是极其影响体验。

而长连接的维护必然需要一套机制来控制。比如 HTTP/1.0 通过在 header 头中添加 Connection:Keep-Alive参数，如果当前请求需要保活则添加该参数作为标识，否则服务端就不会保持该连接的状态，发送完数据之后就关闭连接。HTTP/1.1以后 Keep-Alive 是默认打开的。

Netty 是基于 TCP 协议开发的，在四层协议 TCP 协议的实现中也提供了 keepalive 报文用来探测对端是否可用。TCP 层将在定时时间到后发送相应的 KeepAlive 探针以确定连接可用性。

Netty 中提供了 tcp-keepalive 的设置：

在这里插入图片描述

.childOption(ChannelOption.SO_KEEPALIVE,true) 表示打开 TCP 的 keepAlive 设置。

2 Netty心跳检测机制

Netty 中提供了 IdleStateHandler 类专门用于处理心跳。构造函数如下：

public IdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime,TimeUnit unit) {
    this(false, readerIdleTime, writerIdleTime, allIdleTime, unit);
}

参数说明：

readerIdleTime 隔多久检查一下读事件是否发生，如果 channelRead() 方法超过 readerIdleTime 时间未被调用则会触发超时事件调用 userEventTrigger() 方法
writerIdleTime 隔多久检查一下写事件是否发生，如果 write() 方法超过 writerIdleTime 时间未被调用则会触发超时事件调用 userEventTrigger() 方法；
allIdleTime 隔多久检查读写事件是否发生
unit 时间单位

可以分别控制读，写，读写超时的时间，如果设置为0表示不检测，所以如果全是0，则相当于没添加这个 IdleStateHandler，连接是个普通的短连接。

2.1 代码演示

服务端

public class TestHeartServer {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap=new ServerBootstrap();
            bootstrap.group(bossGroup,workerGroup)
                      .channel(NioServerSocketChannel.class)
                      .option(ChannelOption.SO_BACKLOG,128)
                       .childOption(ChannelOption.SO_KEEPALIVE,true)
                       .handler(new LoggingHandler(LogLevel.INFO))//bossGroup处理handler
                       .childHandler(new ChannelInitializer<SocketChannel>() {//workergroup处理handler
                           @Override
                           protected void initChannel(SocketChannel ch) throws Exception {
                               ChannelPipeline pipeline = ch.pipeline();
                               //每隔5s检查一下是否有读事件发生
                               pipeline.addLast(new IdleStateHandler(5,0,0, TimeUnit.SECONDS));
                               pipeline.addLast(new TestHeartServerHandler());
                           }
                       });
            ChannelFuture channelFuture = bootstrap.bind(9999).sync();
            channelFuture.channel().closeFuture().sync();
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端handler

public class TestHeartServerHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channelActive");
    }
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf=(ByteBuf) msg;
        System.out.println("客户端消息："+buf.toString(StandardCharsets.UTF_8));
        //向客户端发送消息
        //ctx.writeAndFlush(Unpooled.copiedBuffer("heart",StandardCharsets.UTF_8));
    }
    /**
     *如果5s没有读请求，则向客户端发送心跳
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        IdleStateEvent event = (IdleStateEvent) evt;
        switch (event.state()) {
            case READER_IDLE: //读空闲
                //如果5s没有读请求，则向客户端发送心跳
                ctx.writeAndFlush("server send Heartbeat").addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
                break;
            case WRITER_IDLE://写空闲
                break;
            case ALL_IDLE://读写空闲
                break;
        }
    }
}

客户端

public class TestHeartClient {
    public static void main(String[] args) {
        EventLoopGroup eventExecutors=new NioEventLoopGroup();
        try {
            Bootstrap bootstrap=new Bootstrap();
            bootstrap.group(eventExecutors)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            //每隔4s检查一下是否有写事件
                            pipeline.addLast(new IdleStateHandler(0,4,0, TimeUnit.SECONDS));
                            pipeline.addLast(new TestHeartClientHandler());
                        }
                    });
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9999).sync();
            //向服务端发送消息
            channelFuture.channel().writeAndFlush(Unpooled.copiedBuffer("Hello server, i'm online", StandardCharsets.UTF_8));
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            eventExecutors.shutdownGracefully();
        }
    }
}

客户端Handler

public class TestHeartClientHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf=(ByteBuf) msg;
        System.out.println("服务端发送的消息:"+buf.toString(StandardCharsets.UTF_8));
    }
    /**
     *
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        IdleStateEvent event = (IdleStateEvent) evt;
        String eventType = null;
        switch (event.state()) {
            //读空闲
            case READER_IDLE:
                break;
            case WRITER_IDLE://写空闲
                //如果4s没有收到写请求，则向服务端发送心跳请求
                ctx.writeAndFlush(Unpooled.copiedBuffer("client send Heartbeat",StandardCharsets.UTF_8)).addListener(ChannelFutureListener.CLOSE_ON_FAILURE) ;
                break;
            case ALL_IDLE://读写空闲
                break;
        }
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
         cause.printStackTrace();
         ctx.close();
    }
}

解释一下代码的逻辑：

服务端添加了：

Copypipeline.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS));

每隔5s检查一下是否有读事件发生，如果没有就触发 handler 中的 userEventTriggered(ChannelHandlerContext ctx, Object evt)逻辑。

客户端添加了：

Copynew IdleStateHandler(0, 4, 0, TimeUnit.SECONDS)

每隔4s检查一下是否有写事件，如果没有就触发 handler 中的 userEventTriggered(ChannelHandlerContext ctx, Object evt)逻辑。

到此这篇关于Netty中的心跳检测机制详解的文章就介绍到这了,更多相关Netty心跳检测机制内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

您可能感兴趣的文章:

Java使用分布式实现数据库读写分离的示例代码
在MySQL中可以通过配置主从复制来实现数据同步,本文主要介绍了Java使用分布式实现数据库读写分离,具有一定的参考价值,感兴趣的可以了解一下
2024-07-07
JAVA基于数组实现的商品信息查询功能示例
这篇文章主要介绍了JAVA基于数组实现的商品信息查询功能,结合实例形式详细分析了java使用数组存储数据实现的商品信息查询功能相关操作技巧,需要的朋友可以参考下
2019-11-11
Spring Boot基于Active MQ实现整合JMS
这篇文章主要介绍了Spring Boot基于Active MQ实现整合JMS,文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值,需要的朋友可以参考下
2020-07-07
JDK1.8新特性之方法引用 ::和Optional详解
这篇文章主要介绍了JDK1.8新特性之方法引用 ::和Optional,本文通过实例代码给大家介绍的非常详细，对大家的学习或工作具有一定的参考借鉴价值，需要的朋友可以参考下
2021-09-09
SpringBoot ApplicationContextAware拓展接口使用详解
当一个类实现了这个接口（ApplicationContextAware）之后，这个类就可以方便获得ApplicationContext中的所有bean。换句话说，就是这个类可以直接获取spring配置文件中，所有有引用到的bean对象
2023-04-04
Java多线程编程中使用Condition类操作锁的方法详解
Condition是java.util.concurrent.locks包下的类,提供了对线程锁的更精细的控制方法,下面我们就来看一下Java多线程编程中使用Condition类操作锁的方法详解
2016-07-07
Java线程本地变量导致的缓存问题解决方法
使用缓存可以缓解大流量压力,显著提高程序的性能,我们在使用缓存系统时,尤其是大并发情况下,经常会遇到一些疑难杂症,这篇文章主要给大家介绍了关于Java线程本地变量导致的缓存问题的解决方法,需要的朋友可以参考下,
2024-08-08
java类中使用Jfreechart的简单实例
这篇文章介绍了java类中使用Jfreechart的简单实例，有需要的朋友可以参考一下
2013-08-08
java项目中classpath的理解
这篇文章介绍了java项目中classpath的理解，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2021-12-12
详解Java 集合类 List 的那些坑
这篇文章主要介绍了Java 集合类 List 的那些坑，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2020-08-08