Netty心跳检测代码及源码流程

迈不过友情╰ 2022-11-15 11:46 232阅读 0赞

心跳检测是指在TCP长连接中，客户端和服务端定时发送和接受简单数据，确保服务正常，在Netty中，对心跳检测进行了很好的封装，下面我们来看一下心跳检测的实现和源码

# Netty通过什么来实现心跳？ #

IdleStateHandler：Netty是通过IdleStateHandler来实现心跳检测的。

# 怎么使用？ #

## 客户端 ##

public class HeartBeatClient {
        public static void main(String[] args) throws Exception {
    
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group).channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                //加入解码器，接受新的信息时触发
                                pipeline.addLast(new StringDecoder());
                                //加入编码器，当发出信息时候进行编码后再发出
                                pipeline.addLast(new StringEncoder());
                                //加入客服端的业务处理器
                                pipeline.addLast(new HeartBeatClientHandler());
                            }
                        });
    
                System.out.println("netty client start。。");
                //同步连接服务端
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 9000).sync();
                Channel channel = channelFuture.channel();
                while (channel.isActive()) {
                    //每两秒发一次心跳检测
                    Thread.sleep(2000);
    //                每四秒进行一次心跳检测
    //                Thread.sleep(4000);
                    channel.writeAndFlush("心跳检测");
                }
            } finally {
                group.shutdownGracefully();
            }
        }
    }
    
    public 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("服务端关闭连接")) {
                System.out.println(" 服务端关闭连接，客户端也关闭");
                ctx.channel().closeFuture();
            }
        }
    }

## 服务端 ##

public class HeartBeatServer {
        public static void main(String[] args) throws Exception {
            //接受连接线程池
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            //处理业务线程池，默认是cpu凉杯
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ChannelPipeline pipeline = ch.pipeline();
                                //加入解码器，接受新的信息时触发
                                pipeline.addLast(new StringDecoder());
                                //加入编码器，当发出信息时候进行编码后再发出
                                pipeline.addLast(new StringEncoder());
                                //加入心跳检测处理器
                                // readerIdleTime  读超时时间
                                // writerIdleTime  写超时时间
                                // allIdleTime     总的超时时间
    
                                pipeline.addLast(new IdleStateHandler(3, 0, 0, TimeUnit.SECONDS));
                                //加入服务端的业务处理器
                                pipeline.addLast(new HeartBeatServerHandler());
                            }
                        });
    
                System.out.println("netty server start。。");
                //
                ChannelFuture channelFuture = serverBootstrap.bind(9000).sync();
                channelFuture.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }
    
    public class HeartBeatServerHandler extends SimpleChannelInboundHandler<String>  {
        int readIdleTimes = 0;
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
            System.out.println(" ====== > [server] message received : " + msg);
            if ("心跳检测".equals(msg)) {
                ctx.channel().writeAndFlush("Heath ok");
            } else {
                System.out.println(" 其他信息处理 ... ");
            }
        }
    
        /**
         * netty将超时的事务逻辑交由用户自己处理
         * @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:
                    eventType = "读空闲";
                    readIdleTimes++; // 读空闲的计数加1
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    // 不处理
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    // 不处理
                    break;
            }
            System.out.println(ctx.channel().remoteAddress() + "超时事件：" + eventType);
            if (readIdleTimes > 3) {
                System.out.println(" [server]读空闲超过3次，关闭连接，释放更多资源");
                ctx.channel().writeAndFlush("服务端关闭连接");
                ctx.channel().close();
            }
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.err.println("=== " + ctx.channel().remoteAddress() + " is active ===");
        }
    }

# IdleStateHandler源码 #

## 构造方法 ##

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

构造方法没什么特别，主要是对其中一些属性赋值。

## channelActive方法 ##

@Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // This method will be invoked only if this handler was added
            // before channelActive() event is fired.  If a user adds this handler
            // after the channelActive() event, initialize() will be called by beforeAdd().
            initialize(ctx); //进入查看源码
            super.channelActive(ctx);
        }

channelActive中有个initialize(ctx)方法，看一下这个方法

private void initialize(ChannelHandlerContext ctx) {
            // Avoid the case where destroy() is called before scheduling timeouts.
            // See: https://github.com/netty/netty/issues/143
            switch (state) {
                case 1:
                case 2:
                    return;
            }
    
            state = 1;
            initOutputChanged(ctx);
    
            lastReadTime = lastWriteTime = ticksInNanos();  //当前时间
            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);
            }
        }

initialize(ctx)方法主要是对一些属性赋值和判断是否要进行读，写等超时的定时方法，因为我们是测试读超时，那么我们看一下ReaderIdleTimeoutTask的run方法

@Override
            protected void run(ChannelHandlerContext ctx) {
                long nextDelay = readerIdleTimeNanos;
                if (!reading) { //获取是否超时 nextDelay<0 超时 nextDelay>0则不超时
                    nextDelay -= ticksInNanos() - lastReadTime;
                }
    
                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); //交给自己实现的userEventTriggered进行处理
                    } catch (Throwable t) {
                        ctx.fireExceptionCaught(t);
                    }
                } else {
                    // Read occurred before the timeout - set a new timeout with shorter delay. 没超时，则再次调用schedule()方法
                    readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
                }
            }

这里首先会判断是否以前超过了预订的超时时间，若没超时，则将赋值剩余超时时间，继续调用本方法，若超时，则调用我们自己实现的userEventTriggered()方法处理业务逻辑

## 流程图 ##

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NjMwODEw_size_16_color_FFFFFF_t_70][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM4NjMwODEw_size_16_color_FFFFFF_t_70]: /images/20221022/9fc89424498b4479a9ac39dbbe126b2f.png