netty之心跳机制

import org.jboss.netty.bootstrap.ServerBootstrap;import org.jboss.netty.channel.ChannelPipeline;import org.jboss.netty.channel.ChannelPipelineFactory;import org.jboss.netty.channel.Channels;import org.jboss.netty.channel.socket.nio.NioServerSocketChannelFactory;import org.jboss.netty.handler.codec.string.StringDecoder;import org.jboss.netty.handler.codec.string.StringEncoder;import org.jboss.netty.handler.timeout.IdleStateHandler;import org.jboss.netty.util.HashedWheelTimer;import java.net.InetSocketAddress;import java.util.concurrent.ExecutorService;import java.util.concurrent.Executors;public class Server {    public static void main(String[] args) {        //声明服务类        ServerBootstrap serverBootstrap = new ServerBootstrap();        //设定线程池        ExecutorService boss = Executors.newCachedThreadPool();        ExecutorService work = Executors.newCachedThreadPool();        //设置工厂        serverBootstrap.setFactory(new NioServerSocketChannelFactory(boss,work));        //设置管道流        serverBootstrap.setPipelineFactory(new ChannelPipelineFactory() {            @Override            public ChannelPipeline getPipeline() throws Exception {                ChannelPipeline channelPipeline = Channels.pipeline();                //添加处理方式                channelPipeline.addLast("idle",new IdleStateHandler(new HashedWheelTimer(),5,5,10));                channelPipeline.addLast("decode",new StringDecoder());                channelPipeline.addLast("encode",new StringEncoder());                channelPipeline.addLast("server",new ServerHandler());                return channelPipeline;            }        });        //设置端口        serverBootstrap.bind(new InetSocketAddress(9000));    }}

import org.jboss.netty.channel.*;import org.jboss.netty.handler.timeout.IdleState;import org.jboss.netty.handler.timeout.IdleStateEvent;public class ServerHandler extends SimpleChannelHandler {    @Override    public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {        System.out.println("client:"+e.getMessage());        ctx.getChannel().write(e.getMessage());        super.messageReceived(ctx, e);    }    @Override    public void handleUpstream(ChannelHandlerContext ctx, ChannelEvent e) throws Exception {        if (e instanceof IdleStateEvent) {            if (((IdleStateEvent)e).getState() == IdleState.ALL_IDLE) {                ChannelFuture channelFuture = ctx.getChannel().write("Time out,You will close");                channelFuture.addListener(channelFuture1 -> ctx.getChannel().close());            }        } else {            super.handleUpstream(ctx, e);        }    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {        super.exceptionCaught(ctx, e);    }    @Override    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {        super.channelConnected(ctx, e);    }    @Override    public void channelDisconnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {        super.channelDisconnected(ctx, e);    }    @Override    public void channelClosed(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {        super.channelClosed(ctx, e);    }}

import io.netty.bootstrap.ServerBootstrap;import io.netty.channel.*;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.nio.NioServerSocketChannel;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.codec.string.StringEncoder;import io.netty.handler.timeout.IdleStateHandler;public class Server {    public static void main(String[] args) {        //服务类        ServerBootstrap serverBootstrap = new ServerBootstrap();        //声明两个线程池        EventLoopGroup boss = new NioEventLoopGroup();        EventLoopGroup work = new NioEventLoopGroup();        try {            //设置线程组            serverBootstrap.group(boss,work);            //设置服务socket工厂            serverBootstrap.channel(NioServerSocketChannel.class);            //设置管道            serverBootstrap.childHandler(new ChannelInitializer<Channel>() {                protected void initChannel(Channel channel) throws Exception {                    channel.pipeline().addLast(new IdleStateHandler(5,5,10));                    channel.pipeline().addLast(new StringDecoder());                    channel.pipeline().addLast(new StringEncoder());                    channel.pipeline().addLast(new ServerHandler());                }            });            //设置服务器连接数            serverBootstrap.option(ChannelOption.SO_BACKLOG,2048);            //设置tcp延迟状态            serverBootstrap.option(ChannelOption.TCP_NODELAY,true);            //设置激活状态，2小时清除            serverBootstrap.option(ChannelOption.SO_KEEPALIVE,true);            //监听端口            ChannelFuture channelFuture = serverBootstrap.bind(9000);            //等待服务器关闭            channelFuture.channel().closeFuture().sync();        } catch (Exception e) {            e.printStackTrace();        } finally {            //关闭线程池            boss.shutdownGracefully();            work.shutdownGracefully();        }    }}

import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelFutureListener;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.handler.timeout.IdleState;import io.netty.handler.timeout.IdleStateEvent;public class ServerHandler extends SimpleChannelInboundHandler<String> {    //接收消息并处理    protected void messageReceived(ChannelHandlerContext channelHandlerContext, String s) throws Exception {        System.out.println(s);        channelHandlerContext.writeAndFlush("hello client");    }    @Override    public void userEventTriggered(final ChannelHandlerContext ctx,   evt) throws Exception {        if (evt instanceof IdleStateEvent) {            if (((IdleStateEvent)evt).state() == IdleState.ALL_IDLE) {                ChannelFuture channelFuture = ctx.writeAndFlush("Time out,You will close");                channelFuture.addListener(new ChannelFutureListener() {                    public void operationComplete(ChannelFuture channelFuture) throws Exception {                        ctx.channel().close();                    }                });            }        } else {            super.userEventTriggered(ctx, evt);        }    }}