Netty章节五：Netty实现心跳检测

前言：
集群之间的主节点与从节点之间实现数据的最终一致性。节点与节点之间实现数据的异步同步。节点与节点之间怎样才能感知对应节点状态。这就要求节点每隔一段时间定时的发送心跳包去感知对方的服务健康状态。一般在设置几个心跳包之后我们就可以认为对方节点已经挂了，我们就可以将该节点从集群中踢出去。

我们有个疑问，比如说之前的多客户端通信demo，当客户端断开与服务器连接的时候会触发handlerRemoved方法，那么我们就知道该服务的状态了。为什么还需要心跳包去感知呢？

真实情况远比我们想象中的复杂，比如我们的客户端是移动手机并且已经建立好了连接，当打开飞行模式（或者强制关机）的时候我们就无法感知当前连接已经断开了（handlerRemoved不会触发的），

当我们客户端和服务器端进行通信的时候，关闭网络或者打开飞行模式，此时通过handlerAdded方法和handlerRemoved是无法判断服务是否已经宕掉的。那么就引出了本文的内容。

什么是心跳检测？

判断对方（设备，进程或其它网元）是否正常动行，一般采用定时发送简单的通讯包，如果在指定时间段内未收到对方响应，则判断对方已经宕掉。用于检测TCP的异常断开。

基本原因是服务器端不能有效的判断客户端是否在线，也就是说服务器无法区分客户端是长时间在空闲，还是已经掉线的情况。所谓的心跳包就是客户端定时发送简单的信息给服务器端告诉它我还在而已。

代码就是每隔几分钟发送一个固定信息给服务端，服务端收到后回复一个固定信息。如果服务端几分钟内没有收到客户端信息则视客户端断开。比如有些通信软件长时间不使用，要想知道它的状态是在线还是离线就需要心跳包，定时发包收包。

发包方可以是客户也可以是服务端，看哪边实现方便合理。一般是客户端。服务器也可以定时轮询发心跳下去。

一般来说，出于效率的考虑，是由客户端主动向服务器端发包，而不是相反。
在分布式集群部署环境中也经常使用到心跳检测，比如主从服务之间的心跳检查，各master之间的互相检测等等，所以还是非常有实践意义的。

Netty最简单心跳监测实践

服务端代码

服务端主启动类

public class MyServer {
    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new MyServerInitializer());

            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        }catch (Exception e){
            System.out.println(e.getMessage());
        }finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

初始化器 (Initializer)

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        /*
            IdleStateHandler 对空闲检测提供的handler/处理器   检测的是服务器端
            说明：当某个Channel在一定的时间，间隔之内没有进行读、写或读和写的操作时
            就会出发IdleStateEvent事件

            IdleStateHandler(读的时间,写的时间,读写的时间)　默认秒
                读写的时间　读和写空闲触发一个就会触发这个事件
                例：IdleStateHandler(10,10,10)  如果在10秒之内都没有进行读操作，就会触发一个事件，后面两个一样
            IdleStateHandler(读的时间,写的时间,读和写的时间,时间单位<不指定默认是秒>)
         */
        pipeline.addLast(new IdleStateHandler(5,7,10, TimeUnit.SECONDS));
        pipeline.addLast(new MyServerHandler());
    }
}

自定义处理器 (Handler)

public class MyServerHandler extends ChannelInboundHandlerAdapter {
    /**
     * 用户事件被触发，触发某个事件之后就会被调用
     * 它会调研ChannelHandlerContext的fireUserEventTriggered(Object)方法
     * 来将事件转发给ChannelPipeline管道的下一个ChannelInboundHandler对象
     * @param ctx 上下文对象
     * @param evt 事件对象
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        //如果这个事件本身是一个IdleStateEvent类型　表示他是一个空闲状态事件
        if(evt instanceof IdleStateEvent){
            //强制类型转换一下
            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;
                default:
                    break;
            }
            System.out.println(ctx.channel().remoteAddress() + "超时事件：" + eventType);
            ctx.channel().close();
        }
    }
}

客户端代码

客户端主启动类

public class MyChatClient {
    public static void main(String[] args) {

        //事件循环组,只有一个循环组
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new MyChatClientInitializer());
            //与对应的url建立连接通道  .channel 拿到对应的通道对象
            //可以直接与连接该通道的服务交互
            Channel channel = bootstrap.connect("localhost", 8899).sync().channel();

            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
            for (; ;){
                //readLine每次读取一行
                //读取一行数据，回车即读取
                channel.writeAndFlush(br.readLine() + "
");
            }

        }catch (Exception e){
            System.out.println("异常");
            System.out.println(e.getMessage());
        }finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

初始化器 (Initializer)

public class MyChatClientInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));
        //虽然默认StringEncoder默认就是UTF_8，但是最好还是写上
        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
        pipeline.addLast(new MyChatClientHandler());
    }
}

自定义处理器 (Handler)

public class MyChatClientHandler extends SimpleChannelInboundHandler<String> {

    /**
     *
     * @param ctx 上下文请求对象
     * @param msg 表示服务端发来的消息
     * @throws Exception
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        System.out.println(msg);
    }
}

测试

启动服务器和客户端，当5s内客户端没有往服务端写数据，造成了服务器端读空闲，服务器的控制台上打印：

五月 08, 2020 2:09:14 下午 io.netty.handler.logging.LoggingHandler channelRegistered
信息: [id: 0x3cf24080] REGISTERED
五月 08, 2020 2:09:14 下午 io.netty.handler.logging.LoggingHandler bind
信息: [id: 0x3cf24080] BIND: 0.0.0.0/0.0.0.0:8899
五月 08, 2020 2:09:14 下午 io.netty.handler.logging.LoggingHandler channelActive
信息: [id: 0x3cf24080, L:/0:0:0:0:0:0:0:0:8899] ACTIVE
五月 08, 2020 2:09:19 下午 io.netty.handler.logging.LoggingHandler channelRead
信息: [id: 0x3cf24080, L:/0:0:0:0:0:0:0:0:8899] READ: [id: 0x1ce09ceb, L:/127.0.0.1:8899 - R:/127.0.0.1:47456]
五月 08, 2020 2:09:19 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
信息: [id: 0x3cf24080, L:/0:0:0:0:0:0:0:0:8899] READ COMPLETE
/127.0.0.1:47456超时事件：读空闲

修改服务器端Initializer中的代码

pipeline.addLast(new IdleStateHandler(5,3,10, TimeUnit.SECONDS));

重启服务器和客户端，当服务器端3s内没有往客户端写数据，则造成服务器的写空闲。服务器端控制台打印：

五月 08, 2020 2:11:00 下午 io.netty.handler.logging.LoggingHandler channelRegistered
信息: [id: 0xe40d94ba] REGISTERED
五月 08, 2020 2:11:00 下午 io.netty.handler.logging.LoggingHandler bind
信息: [id: 0xe40d94ba] BIND: 0.0.0.0/0.0.0.0:8899
五月 08, 2020 2:11:00 下午 io.netty.handler.logging.LoggingHandler channelActive
信息: [id: 0xe40d94ba, L:/0:0:0:0:0:0:0:0:8899] ACTIVE
五月 08, 2020 2:11:06 下午 io.netty.handler.logging.LoggingHandler channelRead
信息: [id: 0xe40d94ba, L:/0:0:0:0:0:0:0:0:8899] READ: [id: 0x3ee5751e, L:/127.0.0.1:8899 - R:/127.0.0.1:47478]
五月 08, 2020 2:11:06 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
信息: [id: 0xe40d94ba, L:/0:0:0:0:0:0:0:0:8899] READ COMPLETE
/127.0.0.1:47478超时事件：写空闲

修改服务器端Initializer中的代码

 pipeline.addLast(new IdleStateHandler(5,7,4, TimeUnit.SECONDS));

当客户端没有往服务器写数据（造成服务器读事件）和服务端没有往客户端写数据（造成服务器端写事件）的时间达到4s则触发服务端读写空闲。
重启客户端和服务器服务，4s后服务器端控制台打印：

五月 08, 2020 2:12:42 下午 io.netty.handler.logging.LoggingHandler channelRegistered
信息: [id: 0xa6f3a08d] REGISTERED
五月 08, 2020 2:12:42 下午 io.netty.handler.logging.LoggingHandler bind
信息: [id: 0xa6f3a08d] BIND: 0.0.0.0/0.0.0.0:8899
五月 08, 2020 2:12:42 下午 io.netty.handler.logging.LoggingHandler channelActive
信息: [id: 0xa6f3a08d, L:/0:0:0:0:0:0:0:0:8899] ACTIVE
五月 08, 2020 2:13:32 下午 io.netty.handler.logging.LoggingHandler channelRead
信息: [id: 0xa6f3a08d, L:/0:0:0:0:0:0:0:0:8899] READ: [id: 0xec04855e, L:/127.0.0.1:8899 - R:/127.0.0.1:47528]
五月 08, 2020 2:13:32 下午 io.netty.handler.logging.LoggingHandler channelReadComplete
信息: [id: 0xa6f3a08d, L:/0:0:0:0:0:0:0:0:8899] READ COMPLETE
/127.0.0.1:47528超时事件：读写空闲