摘要：使用實(shí)現(xiàn)心跳機(jī)制代碼環(huán)境和具體思路如下使用提供的來檢測(cè)讀寫操作的空閑時(shí)間使用序列化客戶端空閑后向服務(wù)端發(fā)送一個(gè)心跳包服務(wù)端空閑后心跳丟失計(jì)數(shù)器丟失的心跳包數(shù)量當(dāng)丟失的心跳包數(shù)量超過個(gè)時(shí)，主動(dòng)斷開該客戶端的斷開連接后，客戶端之后重新連接代碼已

使用Netty提供的IdleStateHandler來檢測(cè)讀寫操作的空閑時(shí)間

使用Protocol Buffer序列化

客戶端write空閑5s后向服務(wù)端發(fā)送一個(gè)心跳包

服務(wù)端read空閑6s后心跳丟失計(jì)數(shù)器+1（丟失的心跳包數(shù)量）

當(dāng)丟失的心跳包數(shù)量超過3個(gè)時(shí)，主動(dòng)斷開該客戶端的channel

斷開連接后，客戶端10s之后重新連接

option java_outer_classname = "PacketProto";

message Packet {

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

    // 包類型
    required PacketType packetType = 1;
    
    // 數(shù)據(jù)部分（可選，心跳包不包含數(shù)據(jù)部分）
    optional string data = 2;
}

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 之后嘗試重新連接服務(wù)器
        System.out.println("10s 之后嘗試重新連接服務(wù)器...");
        Thread.sleep(10 * 1000);
        Client.doConnect();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            // 不管是讀事件空閑還是寫事件空閑都向服務(wù)器發(fā)送心跳包
            sendHeartbeatPacket(ctx);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("連接出現(xiàn)異常");
    }

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

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());
                        }
                    });

            // 連接服務(wù)器
            doConnect();

            // 模擬不定時(shí)發(fā)送向服務(wù)器發(fā)送數(shù)據(jù)的過程
            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("我是數(shù)據(jù)包（非心跳包） " + num);
                PacketProto.Packet packet = builder.build();
                ch.writeAndFlush(packet);
            }

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

    /**
     * 抽取出該方法 (斷線重連時(shí)使用)
     *
     * @throws InterruptedException
     */
    public static void doConnect() throws InterruptedException {
        ch = bootstrap.connect("127.0.0.1", 20000).sync().channel();
    }
}

public class ServerHeartbeatHandler extends ChannelInboundHandlerAdapter {

    // 心跳丟失計(jì)數(shù)器
    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之后觸發(fā) (心跳包丟失)
            if (counter >= 3) {
                // 連續(xù)丟失3個(gè)心跳包 (斷開連接)
                ctx.channel().close().sync();
                System.out.println("已與Client斷開連接");
            } else {
                counter++;
                System.out.println("丟失了第 " + counter + " 個(gè)心跳包");
            }
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("連接出現(xiàn)異常");
    }

    /**
     * 處理心跳包
     *
     * @param ctx
     * @param packet
     */
    private void handleHeartbreat(ChannelHandlerContext ctx, Packet packet) {
        // 將心跳丟失計(jì)數(shù)器置為0
        counter = 0;
        System.out.println("收到心跳包");
        ReferenceCountUtil.release(packet);
    }

    /**
     * 處理數(shù)據(jù)包
     *
     * @param ctx
     * @param packet
     */
    private void handleData(ChannelHandlerContext ctx, Packet packet) {
        // 將心跳丟失計(jì)數(shù)器置為0
        counter = 0;
        String data = packet.getData();
        System.out.println(data);
        ReferenceCountUtil.release(packet);
    }
}

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();
        }
    }
}