netty心跳

news2024/9/24 15:16:46

为什么要引入心跳机制：

因为网络连接是不可靠的，像在TCP长连接或者webSocket长连接时都会使用心跳机制，即发送特殊的数据包告诉自己的业务没办完，不要关闭连接。

案例要求：

当服务器超过3秒没有读时，就提示读空闲

当服务器超过5秒没有写时，就提示写空闲

当服务器超过7秒没有读写时，就提示读写空闲

Netty中的心跳检测主要是通过IdleStateHandler这个类进行心跳的处理，它可以对一个Channel的读\写设置定时器，当Channel在一定时间间隔内没有数据交互时（即处于Idle状态）就会触发指定的事件

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

readerIdleTimeSeconds：设置读超时时间；
writerIdleTimeSeconds：设置写超时时间；
allIdleTimeSeconds：同时为读或写设置超时时间；

代码示例：

服务端代码：

public class MyServer {

    //编写run方法，处理客户端请求
    public static void main(String[] args)  throws Exception{
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workGroup)
                    .channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,128)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //获取到pipeline
                            ChannelPipeline pipeline = ch.pipeline();
                            //加入一个netty 提供IdleStateHandler
                            /**
                             * 说明:
                             *  1.IdleStateHandler 是netty提供的处理空闲状态的处理量
                             *  2.readerIdleTime 表示多长时间没有读
                             *  3.writerIdleTime 表示多长时间没有写
                             *  4.allIdleTime 表示多长时间没有读写，就会发送一个心跳检测包检测是否连接
                             *  5. unit 时间单位
                             *  6.当IdleStateHandler触发后，就会传递给管道的下一个handle去处理，通过调用触发下一个handle的userEventTiggered
                             *  在该方法中去处理IdleStateHandler(读空闲、写空闲、读写空闲)
                             */
                            pipeline.addLast(new IdleStateHandler(3,5,7, TimeUnit.SECONDS));

                            //加入一个空闲检测进一步处理的handle
                            pipeline.addLast(new MyServerHandle());
                        }
                    });

            //启动服务器
            ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
            channelFuture.channel().closeFuture().sync();

        }finally {
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }


    }
}

服务端处理器：

public class MyServerHandle extends ChannelInboundHandlerAdapter {

    /**
     *
     * @param ctx 上下文
     * @param evt 事件
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {

        if (evt instanceof IdleStateEvent){

            //将evt向下转型IdleStateHandler
            IdleStateEvent event  = (IdleStateEvent)evt;
            String eventType = null;

            switch (event.state()){
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;

            }
            System.out.println(ctx.channel().remoteAddress()+"--超时时间--"+eventType);
            System.out.println("服务器做相应处理");

        }
    }
}

客户端代码：

public class GroupChatClient {

    private final String host;

    private final int port;


    public GroupChatClient(String host, int port) {
        this.port = port;
        this.host = host;
    }

    public void run() throws Exception{

        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap()
                    .group(group)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            //得到通道
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast("decoder",new StringDecoder());
                            pipeline.addLast("encoder",new StringEncoder());
                            //加入自己的业务处理handle
                            pipeline.addLast(new GroupChatClientHandle());

                        }
                    });

            ChannelFuture channelFuture = bootstrap.connect(host, port).sync();

            //得到channel
            Channel channel = channelFuture.channel();
            System.out.println("=================="+channel.localAddress()+"=====================");
            ///客户端需要输入信心，创建一个扫描器
            Scanner scanner = new Scanner(System.in);
            while (scanner.hasNextLine()){
                String msg = scanner.nextLine();
                //通过channel发送到服务器
                channel.writeAndFlush(msg+"\r\n");
            }
        }finally {
            group.shutdownGracefully();
        }




    }

    public static void main(String[] args) throws Exception {
        new GroupChatClient("127.0.0.1",7000).run();
    }

}

客户端处理器：

public class GroupChatClientHandle extends SimpleChannelInboundHandler<String> {

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, String msg) throws Exception {
        System.out.println(msg.trim());
    }
}

启动服务端，再启动客户端，打印如下：