package netty.test1;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.CharsetUtil;import io.netty.util.concurrent.Future;import io.netty.util.concurrent.GenericFutureListener;import org.jboss.netty.channel.ChannelFutureListener;import java.util.concurrent.ScheduledExecutorService;/** * Created by Administrator on 2017/5/17. */@ChannelHandler.Sharablepublic class EchoServerHandler extends ChannelInboundHandlerAdapter { public void channelRegistered(ChannelHandlerContext ctx) throws Exception { System.out.println("注册事件"); } public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { System.out.println("取消注册事件"); } public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("有新客户端连接接入。。。"+ctx.channel().remoteAddress()); } public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println("失去连接"); } public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf in = (ByteBuf) msg; System.out.println("读客户端传入数据="+in.toString(CharsetUtil.UTF_8)); ctx.writeAndFlush(Unpooled.copiedBuffer("channelRead Netty rocks!", CharsetUtil.UTF_8)); //ctx.fireChannelActive(); } public void channelReadComplete(ChannelHandlerContext ctx){ ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(new GenericFutureListener >() { @Override public void operationComplete(Future future) throws Exception { if (future.isSuccess()) { System.out.println("执行成功="+future.isSuccess()); } } }); ctx.writeAndFlush(Unpooled.copiedBuffer("channelReadComplete Netty rocks!", CharsetUtil.UTF_8)).addListener(new GenericFutureListener >() { @Override public void operationComplete(Future future) throws Exception { if (future.isSuccess()) { }else { } } }); } public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { System.out.println("userEventTriggered"); } public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { System.out.println("channelWritabilityChanged"); } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); }}
package netty.test1;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.CharsetUtil;import io.netty.util.ReferenceCountUtil;import io.netty.util.ReferenceCounted;import io.netty.util.concurrent.Future;import io.netty.util.concurrent.GenericFutureListener;/** * Created by Administrator on 2017/5/17. */@ChannelHandler.Sharablepublic class EchoOtherServerHandler extends ChannelInboundHandlerAdapter { public void channelRegistered(ChannelHandlerContext ctx) throws Exception { System.out.println("Other注册事件"); ctx.fireChannelRegistered(); } public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { System.out.println("Other取消注册事件"); ctx.fireChannelUnregistered(); } public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("Other有新客户端连接接入。。。"+ctx.channel().remoteAddress()); ctx.fireChannelActive(); } public void channelInactive(ChannelHandlerContext ctx) throws Exception { System.out.println("Other失去连接"); ctx.fireChannelInactive(); } public void channelRead(ChannelHandlerContext ctx, Object msg) { ByteBuf in = (ByteBuf) msg; System.out.println("Other读客户端传入数据="+in.toString(CharsetUtil.UTF_8)); final ByteBuf byteBuf = Unpooled.copiedBuffer("Other channelRead Netty rocks!", CharsetUtil.UTF_8); ctx.writeAndFlush(byteBuf); ctx.fireChannelRead(msg); //ReferenceCountUtil.release(msg); } public void channelReadComplete(ChannelHandlerContext ctx){ ctx.writeAndFlush(Unpooled.EMPTY_BUFFER).addListener(new GenericFutureListener >() { @Override public void operationComplete(Future future) throws Exception { if (future.isSuccess()) { System.out.println("Other执行成功="+future.isSuccess()); } } }); final ByteBuf byteBuf = Unpooled.copiedBuffer("Other channelReadComplete Netty rocks!", CharsetUtil.UTF_8); ctx.writeAndFlush(byteBuf).addListener(new GenericFutureListener >() { @Override public void operationComplete(Future future) throws Exception { if (future.isSuccess()) { }else { } ReferenceCountUtil.release(byteBuf); } }); ctx.fireChannelReadComplete(); } public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception { System.out.println("Other userEventTriggered"); } public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception { System.out.println("Other channelWritabilityChanged"); } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); }}
package netty.test1;import io.netty.buffer.ByteBuf;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelOutboundHandler;import io.netty.channel.ChannelOutboundHandlerAdapter;import io.netty.channel.ChannelPromise;import java.nio.charset.Charset;/** * Created by Administrator on 2017/6/8. */public class EchoServerOutHandler extends ChannelOutboundHandlerAdapter { public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { System.out.println("EchoServerOutHandler write "+((ByteBuf)msg).toString(Charset.defaultCharset())); ctx.write(msg, promise); }}
package netty.test1;import io.netty.bootstrap.ServerBootstrap;import io.netty.buffer.ByteBufAllocator;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.ChannelPipeline;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioServerSocketChannel;import java.net.InetSocketAddress;/** * Created by Administrator on 2017/5/17. */public class EchoServer { private final int port; public EchoServer(int port) { this.port = port; } public static void main(String[] args) throws InterruptedException { int port = 8080; ByteBufAllocator allocator; new EchoServer(port).start(); } public void start() throws InterruptedException { final EchoServerHandler echoServerHandler = new EchoServerHandler(); final EchoOtherServerHandler echoOtherServerHandler = new EchoOtherServerHandler(); final EchoServerOutHandler echoServerOutHandler = new EchoServerOutHandler(); EventLoopGroup eventLoopGroup = new NioEventLoopGroup(); try { ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(eventLoopGroup). channel(NioServerSocketChannel.class).//指定channel使用Nio传输 localAddress(new InetSocketAddress(port)).//执行端口设置套接字地址 childHandler(new ChannelInitializer () {//添加echoServerHandler到Channel的channelpipeline上 @Override protected void initChannel(SocketChannel socketChannel) throws Exception { ChannelPipeline channelPipeline = socketChannel.pipeline(); channelPipeline.addFirst(echoOtherServerHandler); channelPipeline.addFirst(echoServerOutHandler); channelPipeline.addLast(echoServerHandler); } }); ChannelFuture f = serverBootstrap.bind().sync();//异步绑定服务器，调用sync()方法阻塞等待直到绑定完成 f.channel().closeFuture().sync();//获得Channel的closefutrue，并且阻塞当前线程直到它完成 } catch (InterruptedException e) { eventLoopGroup.shutdownGracefully().sync(); } }}
package netty.test1;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandler;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.channel.SimpleChannelInboundHandler;import io.netty.util.CharsetUtil;import io.netty.util.concurrent.Future;import io.netty.util.concurrent.GenericFutureListener;import org.jboss.netty.channel.ChannelFutureListener;/** * Created by Administrator on 2017/5/17. */@ChannelHandler.Sharablepublic class EchoClientHandler extends SimpleChannelInboundHandler{ public void channelActive(ChannelHandlerContext ctx) throws Exception { ctx.writeAndFlush(Unpooled.copiedBuffer("Netty rocks!", CharsetUtil.UTF_8)); } public void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) { ByteBuf in = msg; System.out.println("读取服务端channelRead0="+in.toString(CharsetUtil.UTF_8)); } public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { cause.printStackTrace(); ctx.close(); }}
package netty.test1;import io.netty.bootstrap.Bootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.oio.OioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;import io.netty.channel.socket.oio.OioSocketChannel;import io.netty.util.concurrent.Future;import io.netty.util.concurrent.GenericFutureListener;import java.net.InetSocketAddress;import java.nio.ByteBuffer;import java.nio.charset.Charset;/** * Created by Administrator on 2017/5/17. */public class EchoClient { private final int port; public EchoClient(int port) { this.port = port; } public static void main(String[] argsw) { ByteBuffer byteBuffer; String host = "127.0.0.1"; int port = 8080; try { new EchoClient(8080).start(); } catch (Exception e) { e.printStackTrace(); } } public void start() throws Exception { final EchoClientHandler clientHandler = new EchoClientHandler(); EventLoopGroup eventLoopGroup = new OioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(eventLoopGroup). channel(OioSocketChannel.class). remoteAddress(new InetSocketAddress("localhost",port)). handler(new ChannelInitializer () { @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(clientHandler); } }); //异步连接远程服务，连接远程服务成功后，输出"已经连接到服务器！" final ChannelFuture f = b.connect(); f.addListener(new GenericFutureListener >() { @Override public void operationComplete(Future future) throws Exception { if (future.isSuccess()) { System.out.println("已经连接到服务器！"); ByteBuf byteBuf = Unpooled.copiedBuffer("创建ByteBuf", Charset.defaultCharset()); ChannelFuture channelFuture = f.channel().writeAndFlush(byteBuf); }else { Throwable throwable = future.cause(); throwable.printStackTrace(); } } }); f.channel().closeFuture().sync(); } catch (InterruptedException e) { eventLoopGroup.shutdownGracefully().sync(); } }} 服务端日志：
Other注册事件
注册事件Other有新客户端连接接入。。。/127.0.0.1:58678有新客户端连接接入。。。/127.0.0.1:58678Other读客户端传入数据=创建ByteBufNetty rocks!EchoServerOutHandler write Other channelRead Netty rocks!读客户端传入数据=创建ByteBufNetty rocks!EchoServerOutHandler write channelRead Netty rocks!EchoServerOutHandler write Other执行成功=trueEchoServerOutHandler write Other channelReadComplete Netty rocks!EchoServerOutHandler write 执行成功=trueEchoServerOutHandler write channelReadComplete Netty rocks!------------------------------------------------------------------------------------------------
客户端日志：
已经连接到服务器！
读取服务端channelRead0=Other channelRead Netty rocks!读取服务端channelRead0=channelRead Netty rocks!读取服务端channelRead0=Other channelReadComplete Netty rocks!读取服务端channelRead0=channelReadComplete Netty rocks!