标签:ada pip ring 客户端 读写 code 一个 end body
package com.hjp.netty.netty; import io.netty.bootstrap.ServerBootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioServerSocketChannel; public class TimeServer { public void bind(int port) throws Exception { //配置服务端的线程组,两个线程组,一个用于服务端接收客户端连接,另一个进行SocketChannel的网络读写 EventLoopGroup bossGroup = new NioEventLoopGroup(); EventLoopGroup workerGroup = new NioEventLoopGroup(); try { //ServerBootstrap用于启动NIO服务端的辅助启动类,目的是降低服务端开发复杂度 ServerBootstrap bootstrap = new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .childHandler(new ChildChannelHandler()); //绑定端口,同步等待成功 ChannelFuture future=bootstrap.bind(port).sync(); //等待服务端监听端口关闭 future.channel().closeFuture().sync(); }finally { //优雅退出,释放线程资源 bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } } private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{ protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(new TimeServerHandler()); } } public static void main(String[] args)throws Exception{ int port=8080; if(args!=null&&args.length>0){ try { port=Integer.valueOf(args[0]); }catch (NumberFormatException e){ } } new TimeServer().bind(port); } }
package com.hjp.netty.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import java.util.Date; public class TimeServerHandler extends ChannelHandlerAdapter { @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf byteBuf= (ByteBuf) msg; byte[] request=new byte[byteBuf.readableBytes()]; byteBuf.readBytes(request); String body=new String(request,"UTF-8"); System.out.println("The time server receive order : "+body); String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new Date().toString():"BAD ORDER"; ByteBuf response= Unpooled.copiedBuffer(currentTime.getBytes()); ctx.write(response); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.flush(); } }
package com.hjp.netty.netty; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelInitializer; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; public class TimeClient { public void connect(String host, int port) throws Exception { //配置客户端NIO线程组 EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap bootstrap = new Bootstrap(); bootstrap.group(group).channel(NioSocketChannel.class) .option(ChannelOption.TCP_NODELAY, true) .handler(new ChannelInitializer<SocketChannel>() { protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(new TimeClientHandler()); } }); //发起异步连接操作,同步等待连接成功 ChannelFuture future=bootstrap.connect(host,port).sync(); //等待客户端链路关闭 future.channel().close().sync(); }finally { //优雅退出,释放NIO线程组 group.shutdownGracefully(); } } public static void main(String[] args)throws Exception{ int port=8080; if(args!=null&&args.length>0){ try { port=Integer.valueOf(port); }catch (NumberFormatException e){ } } new TimeClient().connect("127.0.0.1",port); } }
package com.hjp.netty.netty; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import java.util.logging.Logger; public class TimeClientHandler extends ChannelHandlerAdapter { private static final Logger logger = Logger.getLogger(TimeClientHandler.class.getName()); private final ByteBuf firstMessage; public TimeClientHandler() { byte[] request = "QUERY TIME ORDER".getBytes(); firstMessage = Unpooled.buffer(request.length); firstMessage.writeBytes(request); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { logger.warning("Unexpected exception from downstream : "+cause.getMessage()); ctx.close(); } @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { ctx.writeAndFlush(firstMessage); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf byteBuf= (ByteBuf) msg; byte[] request=new byte[byteBuf.readableBytes()]; byteBuf.readBytes(request); String body=new String(request,"UTF-8"); System.out.println("Now is "+body); } }
标签:ada pip ring 客户端 读写 code 一个 end body
原文地址:http://www.cnblogs.com/hujiapeng/p/7233952.html