Netty实现心跳检测与断线重连

RobinQu 发布于2019-08-14 15:06 / 2528人阅读

摘要：使用实现心跳机制代码环境和具体思路如下使用提供的来检测读写操作的空闲时间使用序列化客户端空闲后向服务端发送一个心跳包服务端空闲后心跳丢失计数器丢失的心跳包数量当丢失的心跳包数量超过个时，主动断开该客户端的断开连接后，客户端之后重新连接代码已

使用Netty实现心跳机制 代码环境：JDK1.8和Netty4.x 具体思路如下：

使用Netty提供的IdleStateHandler来检测读写操作的空闲时间

使用Protocol Buffer序列化

客户端write空闲5s后向服务端发送一个心跳包

服务端read空闲6s后心跳丢失计数器+1（丢失的心跳包数量）

当丢失的心跳包数量超过3个时，主动断开该客户端的channel

断开连接后，客户端10s之后重新连接

代码已上传至GitHub：完整代码地址 代码实现： 数据包结构（proto文件）

option java_outer_classname = "PacketProto";

message Packet {

    // 包的类型
    enum PacketType {
        // 心跳包
        HEARTBEAT = 1;
        // 非心跳包
        DATA = 2;
    }

    // 包类型
    required PacketType packetType = 1;
    
    // 数据部分（可选，心跳包不包含数据部分）
    optional string data = 2;
}

ClientHeartbeatHandler类

public class ClientHeartbeatHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("--- Server is active ---");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("--- Server is inactive ---");

        // 10s 之后尝试重新连接服务器
        System.out.println("10s 之后尝试重新连接服务器...");
        Thread.sleep(10 * 1000);
        Client.doConnect();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            // 不管是读事件空闲还是写事件空闲都向服务器发送心跳包
            sendHeartbeatPacket(ctx);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("连接出现异常");
    }

    /**
     * 发送心跳包
     *
     * @param ctx
     */
    private void sendHeartbeatPacket(ChannelHandlerContext ctx) {
        Packet.Builder builder = newBuilder();
        builder.setPacketType(Packet.PacketType.HEARTBEAT);
        Packet packet = builder.build();
        ctx.writeAndFlush(packet);
    }
}

Client类

public class Client {

    private static Channel ch;
    private static Bootstrap bootstrap;

    public static void main(String[] args) {
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            bootstrap = new Bootstrap();
            bootstrap
                    .group(workGroup)
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                            pipeline.addLast(new ProtobufEncoder());
                            pipeline.addLast(new IdleStateHandler(0, 5, 0));
                            pipeline.addLast(new ClientHeartbeatHandler());
                        }
                    });

            // 连接服务器
            doConnect();

            // 模拟不定时发送向服务器发送数据的过程
            Random random = new Random();
            while (true) {
                int num = random.nextInt(21);
                Thread.sleep(num * 1000);
                PacketProto.Packet.Builder builder = newBuilder();
                builder.setPacketType(PacketProto.Packet.PacketType.DATA);
                builder.setData("我是数据包（非心跳包） " + num);
                PacketProto.Packet packet = builder.build();
                ch.writeAndFlush(packet);
            }

        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            workGroup.shutdownGracefully();
        }
    }

    /**
     * 抽取出该方法 (断线重连时使用)
     *
     * @throws InterruptedException
     */
    public static void doConnect() throws InterruptedException {
        ch = bootstrap.connect("127.0.0.1", 20000).sync().channel();
    }
}

ServerHeartbeatHandler类

public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {

    // 心跳丢失计数器
    private int counter;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("--- Client is active ---");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("--- Client is inactive ---");
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        // 判断接收到的包类型
        if (msg instanceof Packet) {
            Packet packet = (Packet) msg;

            switch (packet.getPacketType()) {
                case HEARTBEAT:
                    handleHeartbreat(ctx, packet);
                    break;

                case DATA:
                    handleData(ctx, packet);
                    break;

                default:
                    break;
            }
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            // 空闲6s之后触发 (心跳包丢失)
            if (counter >= 3) {
                // 连续丢失3个心跳包 (断开连接)
                ctx.channel().close().sync();
                System.out.println("已与Client断开连接");
            } else {
                counter++;
                System.out.println("丢失了第 " + counter + " 个心跳包");
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("连接出现异常");
    }

    /**
     * 处理心跳包
     *
     * @param ctx
     * @param packet
     */
    private void handleHeartbreat(ChannelHandlerContext ctx, Packet packet) {
        // 将心跳丢失计数器置为0
        counter = 0;
        System.out.println("收到心跳包");
        ReferenceCountUtil.release(packet);
    }

    /**
     * 处理数据包
     *
     * @param ctx
     * @param packet
     */
    private void handleData(ChannelHandlerContext ctx, Packet packet) {
        // 将心跳丢失计数器置为0
        counter = 0;
        String data = packet.getData();
        System.out.println(data);
        ReferenceCountUtil.release(packet);
    }
}

Server类

public class Server {
    public static void main(String[] args) {
        NioEventLoopGroup acceptorGroup = new NioEventLoopGroup(1);
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap
                    .group(acceptorGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new ProtobufVarint32FrameDecoder());
                            pipeline.addLast(new ProtobufDecoder(PacketProto.Packet.getDefaultInstance()));
                            pipeline.addLast(new IdleStateHandler(6, 0, 0));
                            pipeline.addLast(new ServerHeartbeatHandler());
                        }
                    });
            Channel ch = bootstrap.bind(20000).sync().channel();
            System.out.println("Server has started...");
            ch.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            acceptorGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

GPU云服务器云服务器 php 实现心跳检测 webrtc断线重连 python断线重连 webrtc 断线重连

文章版权归作者所有，未经允许请勿转载,若此文章存在违规行为，您可以联系管理员删除。

转载请注明本文地址：https://www.ucloud.cn/yun/65289.html

浅析 Netty 实现心跳机制与断线重连

摘要：基础何为心跳顾名思义所谓心跳即在长连接中客户端和服务器之间定期发送的一种特殊的数据包通知对方自己还在线以确保连接的有效性为什么需要心跳因为网络的不可靠性有可能在保持长连接的过程中由于某些突发情况例如网线被拔出突然掉电等会造成服务器和客户端的基础何为心跳顾名思义, 所谓心跳, 即在 TCP 长连接中, 客户端和服务器之间定期发送的一种特殊的数据包, 通知对方自己还在线, 以确保 ...

waterc 2019-08-14 14:42 评论0 收藏0
使用Netty，我们到底在开发些什么？

摘要：比如面向连接的功能包发送接收数量包发送接收速率错误计数连接重连次数调用延迟连接状态等。你要处理的，就是心跳超时的逻辑，比如延迟重连。发生异常后，可以根据不同的类型选择断线重连比如一些二进制协议的编解码紊乱问题，或者调度到其他节点。在java界，netty无疑是开发网络应用的拿手菜。你不需要太多关注复杂的nio模型和底层网络的细节，使用其丰富的接口，可以很容易的实现复杂的通讯功能。和...

DesGemini 2019-08-16 15:23 评论0 收藏0
使用Netty，我们到底在开发些什么？

摘要：比如面向连接的功能包发送接收数量包发送接收速率错误计数连接重连次数调用延迟连接状态等。你要处理的，就是心跳超时的逻辑，比如延迟重连。发生异常后，可以根据不同的类型选择断线重连比如一些二进制协议的编解码紊乱问题，或者调度到其他节点。在java界，netty无疑是开发网络应用的拿手菜。你不需要太多关注复杂的nio模型和底层网络的细节，使用其丰富的接口，可以很容易的实现复杂的通讯功能。和...

MSchumi 2019-06-24 18:21 评论0 收藏0
长连接的心跳及重连设计

摘要：超过后则认为服务端出现故障，需要重连。同时在每次心跳时候都用当前时间和之前服务端响应绑定到上的时间相减判断是否需要重连即可。客户端检测到某个服务端迟迟没有响应心跳也能重连获取一个新的连接。 showImg(https://segmentfault.com/img/remote/1460000017987884?w=800&h=536); 前言说道心跳这个词大家都不陌生，当然不是指男女...

dreamGong 2019-08-16 14:41 评论0 收藏0