【Netty】三、Netty心跳检测与断线重连

﹏ヽ暗。殇╰゛Y 2023-10-09 15:02 10阅读 0赞

#### Netty心跳检测 ####

*  一、Netty心跳检测与断线重连案例
 *   *  客户端代码
     *   *  NettyClient
         *  NettyClientHandler
     *  服务端代码
     *   *  NettyServer
         *  NettyServerChannelInitializer
         *  NettyServerHandler
     *  测试

## 一、Netty心跳检测与断线重连案例 ##

需求：  
1、客户端利用空闲状态给服务端发送心跳ping命令，保持长连接不被关闭；  
2、服务端如果超过指定的时间没有收到客户端心跳，则关闭连接；  
3、服务端关闭连接触发客户端的channelInactive方法，在此方法中进行重连；

### 客户端代码 ###

#### NettyClient ####

import com.mytest.heartbeat.handler.NettyClientHandler;
    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.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    import io.netty.handler.timeout.IdleStateHandler;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * 心跳检测服务端 对应的服务端在netty-server heartbeat 包下的NettyClient
     */
    public class NettyClient {
        
    
        private static Bootstrap bootstrap = new Bootstrap();
    
        public static void main(String[] args) throws InterruptedException {
        
    
    
            try {
        
                NioEventLoopGroup group = new NioEventLoopGroup(8);
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<NioSocketChannel>() {
        
                            @Override
                            protected void initChannel(NioSocketChannel ch) throws Exception {
        
                                ch.pipeline()
                                        //空闲状态的handler
                                        .addLast(new IdleStateHandler(0, 5, 0, TimeUnit.SECONDS))
                                        .addLast(new ObjectEncoder())
                                        .addLast(new ObjectDecoder((s) -> {
        
                                            return String.class;
                                        }))
                                        .addLast(new NettyClientHandler());
                            }
                        });
    
                //连接
                connect();
            } catch (Exception e) {
        
                e.printStackTrace();
            }
        }
    
        /**
         * 重连方法
         */
        public static void connect() {
        
            try {
        
                ChannelFuture future = bootstrap.connect("127.0.0.1", 8787).sync()
                        .addListener(new ChannelFutureListener() {
        
                            @Override
                            public void operationComplete(ChannelFuture ch) throws Exception {
        
                                if (!ch.isSuccess()) {
        
                                    ch.channel().close();
                                    final EventLoop loop = ch.channel().eventLoop();
                                    loop.schedule(new Runnable() {
        
                                        @Override
                                        public void run() {
        
                                            System.err.println("服务端链接不上，开始重连操作...");
                                            //重连
                                            connect();
                                        }
                                    }, 3L, TimeUnit.SECONDS);
                                } else {
        
                                    ch.channel().writeAndFlush("ping");
                                    System.out.println("服务端链接成功...");
                                }
                            }
                        });
            } catch (Exception e) {
        
                System.out.println(e.getMessage());
                try {
        
                    Thread.sleep(3000L);
                } catch (InterruptedException ex) {
        
                    ex.printStackTrace();
                }
                //再重连
                connect();
            }
        }
    }

#### NettyClientHandler ####

import com.mytest.heartbeat.NettyClient;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    
    import java.util.concurrent.TimeUnit;
    
    public class  NettyClientHandler extends ChannelInboundHandlerAdapter {
        
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            System.out.println("连接激活 == " + ctx.channel().id());
        }
    
        @Override
        public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        
            System.out.println("断线了......" + ctx.channel().id());
            ctx.channel().eventLoop().schedule(() -> {
        
                System.out.println("断线重连......");
                //重连
                NettyClient.connect();
            }, 3L, TimeUnit.SECONDS);
        }
    
        /**
         * 用户事件的回调方法
         *
         * @param ctx
         * @param evt
         * @throws Exception
         */
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        
            //如果是空闲状态事件
            if (evt instanceof IdleStateEvent) {
        
                if (((IdleStateEvent) evt).state() == IdleState.WRITER_IDLE) {
        
                    System.out.println("空闲"  + ctx.channel().id());
                    //发送ping 保持心跳链接
                    ctx.writeAndFlush("ping");
                }
            }else {
        
                userEventTriggered(ctx,evt);
            }
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        
            cause.printStackTrace();
            ctx.close();
        }
    }

### 服务端代码 ###

#### NettyServer ####

import com.mytest.heartbeat.handler.NettyServerChannelInitializer;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.ServerSocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    
    /**
     * NettyServer 心跳检测服务端 
     *
     * Netty心跳检测与断线重连
     * 需求：
     * 1、客户端利用空闲状态给服务端发送心跳ping命令，保持长连接不被关闭；
     * 2、服务端如果超过指定的时间没有收到客户端心跳，则关闭连接；
     * 3、服务端关闭连接触发客户端的channelInactive方法，在此方法中进行重连；
     */
    public class NettyServer {
        
    
        public static final int port = 8787;
    
        public static void main(String[] args) {
        
            NettyServer nettyServer = new NettyServer();
            nettyServer.run();
        }
    
    
        private void run() {
        
            NioEventLoopGroup boss = new NioEventLoopGroup(1);
            NioEventLoopGroup work = new NioEventLoopGroup();
            try {
        
                ServerBootstrap bootstrap = new ServerBootstrap().group(boss, work);
                bootstrap.channel(NioServerSocketChannel.class)
                         //这个处理器可以不写
                        .handler(new ChannelInitializer<ServerSocketChannel>() {
        
                            @Override
                            protected void initChannel(ServerSocketChannel ch) throws Exception {
        
                                System.out.println("服务正在启动中......");
                            }
                        })
                        //业务处理
                        .childHandler(NettyServerChannelInitializer.INSTANCE);
    
                ChannelFuture future = bootstrap.bind(port).sync();
    
                future.addListener(f -> {
        
                    if (future.isSuccess()) {
        
                        System.out.println("服务启动成功");
                    } else {
        
                        System.out.println("服务启动失败");
                    }
                });
                future.channel().closeFuture().sync();
    
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } finally {
        
                boss.shutdownGracefully();
                work.shutdownGracefully();
            }
        }
    }

#### NettyServerChannelInitializer ####

自定义的Channe（网络传输操作的接口）  
**ChannelPipeline提供了一个容器给ChannelHandler链并提供了一个API 用于管理沿着链入站和出站事件的流动，每个Channel都有自己的ChannelPipeline，ChannelHandler**

import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.serialization.ObjectDecoder;
    import io.netty.handler.codec.serialization.ObjectEncoder;
    import io.netty.handler.timeout.IdleStateHandler;
    
    import java.util.concurrent.TimeUnit;
    
    @ChannelHandler.Sharable
    public class NettyServerChannelInitializer extends ChannelInitializer<NioSocketChannel> {
        
    
        public static final NettyServerChannelInitializer INSTANCE = new NettyServerChannelInitializer();
    
    //NioSocketChannel，异步的客户端TCP Socket连接
    
        @Override
        protected void initChannel(NioSocketChannel ch) throws Exception {
        
            ch.pipeline()
                    //空闲状态的处理器
                    .addLast(new IdleStateHandler(10, 0, 0, TimeUnit.SECONDS))
                    .addLast(new ObjectEncoder())
                    .addLast(new ObjectDecoder((s) -> {
        
                        return String.class;
                    }))
                    .addLast(NettyServerHandler.INSTANCE);
        }
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        
            System.out.println("有新连接加入了++++......" + ctx.channel().id());
        }
    }

#### NettyServerHandler ####

import io.netty.channel.ChannelHandler;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.handler.timeout.IdleState;
    import io.netty.handler.timeout.IdleStateEvent;
    import io.netty.util.ReferenceCountUtil;
    
    @ChannelHandler.Sharable
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
        
    
        public static final NettyServerHandler INSTANCE = new NettyServerHandler();
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
        
            super.channelActive(ctx);
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        
            try {
        
                System.out.println("接收到的信息：" + msg);
            } finally {
        
                ReferenceCountUtil.release(msg);
            }
        }
    
        @Override
        public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        
    
            //空闲状态的事件
            if (evt instanceof IdleStateEvent) {
        
                IdleStateEvent event = (IdleStateEvent) evt;
                System.out.println(((IdleStateEvent) event).state() + ">>>" + ctx.channel().id());
    
                //已经10秒钟没有读时间了
                if (event.state().equals(IdleState.READER_IDLE)){
        
                    // 心跳包丢失，10秒没有收到客户端心跳 (断开连接)
                    ctx.channel().close().sync();
                    System.out.println("已与 "+ctx.channel().remoteAddress()+" 断开连接");
                }
            }
        }
    }

### 测试 ###

![在这里插入图片描述][2e886785a29545868d6228d1e597ffb7.png]  
![在这里插入图片描述][d8f4d02e1f524117a6ab9ef3b6b67cf0.png]

[2e886785a29545868d6228d1e597ffb7.png]: https://img-blog.csdnimg.cn/2e886785a29545868d6228d1e597ffb7.png
[d8f4d02e1f524117a6ab9ef3b6b67cf0.png]: https://img-blog.csdnimg.cn/d8f4d02e1f524117a6ab9ef3b6b67cf0.png