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, Object 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); ???????} ???}}