netty 心跳检测

心跳检测在很多分布式应用中都存在 mongodb redis ...

服务器端代码

• 1，MyServer.java

public class MyServer {

    public static void main(String... arg) throws Exception {

        EventLoopGroup boss = new NioEventLoopGroup();
        EventLoopGroup workGroup = new NioEventLoopGroup();

        try {

            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boss, workGroup).channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ServerInitializer());

            ChannelFuture channelFuture = bootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();

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


    }
}

• 2,ServerInitializer.java

public class ServerInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();

        //心跳handler 超时时间 读 15秒  写30秒  读写50
        pipeline.addLast(new IdleStateHandler(15,30,50, TimeUnit.SECONDS));

        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));

        pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
        pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));

        pipeline.addLast(new MyServerHandler());


    }
}

• 3， MyServerHandler.java

public class MyServerHandler extends SimpleChannelInboundHandler<String> {

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        Channel channel = ctx.channel();
        String ip = channel.remoteAddress().toString();

        System.out.println("add ip:    " + ip);
        channel.writeAndFlush("【服务器】"+ LocalDateTime.now()+"\n");
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        String ip = ctx.channel().remoteAddress().toString();

        System.out.println("remove ip:    " + ip);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            String eventType = null;
            switch (event.state()) {
                case ALL_IDLE:
                    eventType = "读写空闲";
                    break;
                case WRITER_IDLE:
                    eventType = "写空闲";
                    break;
                case READER_IDLE:
                    eventType = "读空闲";
                    break;
            }

            System.out.println(ctx.channel().remoteAddress()+" 超时时间 "+eventType);


        }
    }


    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        Channel channel = ctx.channel();
        String ip = channel.remoteAddress().toString();
        System.out.println(ip+" ---------- "+msg);
        channel.writeAndFlush("【服务器】"+ LocalDateTime.now()+"\n");
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
       ctx.channel().close();
    }
}

客户端

• 1，MyClient.java

public class MyClient {
    public static void main(String... arg) throws Exception {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

        try {

            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class)
                    .handler(new ClientInitializer());

            Channel channel = bootstrap.connect("127.0.0.1",8899).sync().channel();

            BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
            while (true){
                channel.writeAndFlush(br.readLine()+"\r\n");
            }

        } finally {
            eventLoopGroup.shutdownGracefully();
        }

    }

}

• 2, ClientInitializer.java

public class ClientInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();


        pipeline.addLast(new DelimiterBasedFrameDecoder(4096, Delimiters.lineDelimiter()));

        pipeline.addLast(new StringDecoder());
        pipeline.addLast(new StringEncoder());

        pipeline.addLast(new ClientHander());


    }
}

• 3,ClientHander.java

public class ClientHander extends SimpleChannelInboundHandler<String> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {

        System.out.println(" 接收到数据 " + msg);

    }
}