package com.zhaowb.netty.ch5_1;import io.netty.bootstrap.ServerBootstrap;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;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;import io.netty.handler.codec.DelimiterBasedFrameDecoder;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.logging.LogLevel;import io.netty.handler.logging.LoggingHandler;public class EchoServer { ???public void bind(int port) throws Exception { ???????// 配置 服务端的 NIO 线程组 ???????EventLoopGroup bossGroup = new NioEventLoopGroup(); ???????EventLoopGroup workerGroup = new NioEventLoopGroup(); ???????try { ???????????ServerBootstrap b = new ServerBootstrap(); ???????????b.group(bossGroup, workerGroup) ???????????????????.channel(NioServerSocketChannel.class) ???????????????????.option(ChannelOption.SO_BACKLOG, 1024) ???????????????????.handler(new LoggingHandler(LogLevel.INFO)) ???????????????????.childHandler(new ChannelInitializer<SocketChannel>() { ???????????????????????@Override ???????????????????????protected void initChannel(SocketChannel ch) throws Exception { ???????????????????????????ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());// 创建分隔符对象 ByteBuf 使用 $_ 作为分隔符。 ???????????????????????????// 第一个参数为单条消息的最大长度，当达到该长度后仍然没有查找到分隔符，就抛出 TooLongFrameException 异常 ???????????????????????????// 防止由于异常码流缺失分隔符导致的内存溢出，这是 netty 解码器的可靠性保护;第二个参数是分隔符缓冲对象。 ???????????????????????????ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter)); ???????????????????????????ch.pipeline().addLast(new StringDecoder()); ???????????????????????????ch.pipeline().addLast(new EchoServerHandler()); ???????????????????????} ???????????????????}); ???????????// 绑定端口，同步等待成功。 ???????????ChannelFuture f = b.bind(port).sync(); ???????????// 等待服务端监听端口关闭。 ???????????f.channel().closeFuture().sync(); ???????} catch (Exception e) { ???????????// 退出，释放线程池资源 ???????????bossGroup.shutdownGracefully(); ???????????workerGroup.shutdownGracefully(); ???????} ???} ???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 EchoServer().bind(port); ???}}

package com.zhaowb.netty.ch5_1;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandlerAdapter;import io.netty.channel.ChannelHandlerContext;public class EchoServerHandler extends ChannelHandlerAdapter { ???int counter = 0; ???@Override ???public void channelActive(ChannelHandlerContext ctx) throws Exception { ???????System.out.println("链接成功！"); ???} ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????// 当读取信息时，先经过 DelimiterBasedFrameDecoder 去掉结束分隔符，然后再使用 StringDecoder 将 ByteBuf 解码成字符串/ ???????// 然后 EchoServerHandler 收到的数据就是解码后的字符串对象 ???????String body = (String) msg; ???????System.out.println("This is " + ++counter + " times receive client : [" + body + "]"); ???????body += "$_"; // 由于 设置 DelimiterBasedFrameDecoder 过滤掉了分隔符，返回给客户端时需要在消息的尾部拼接 "$_" ???????ByteBuf echo = Unpooled.copiedBuffer(body.getBytes()); ???????ctx.writeAndFlush(echo); ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close();// 发生异常，关闭链路 ???}}

package com.zhaowb.netty.ch5_1;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.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;import io.netty.handler.codec.DelimiterBasedFrameDecoder;import io.netty.handler.codec.string.StringDecoder;import io.netty.handler.logging.LogLevel;import io.netty.handler.logging.LoggingHandler;public class EchoClient { ???public void connect(int port, String host) throws Exception { ???????//配置客户端 NIO 线程组 ???????EventLoopGroup group = new NioEventLoopGroup(); ???????try { ???????????Bootstrap b = new Bootstrap(); ???????????b.group(group).channel(NioSocketChannel.class) ???????????????????.option(ChannelOption.TCP_NODELAY, true) ???????????????????.handler(new ChannelInitializer<SocketChannel>() { ???????????????????????@Override ???????????????????????protected void initChannel(SocketChannel ch) throws Exception { ???????????????????????????ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());// 创建分隔符对象 ByteBuf 使用 $_ 作为分隔符。 ???????????????????????????ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter)); ???????????????????????????ch.pipeline().addLast(new StringDecoder()); ???????????????????????????ch.pipeline().addLast(new EchoClientHandler()); ???????????????????????} ???????????????????}); ???????????// 发起异步连接操作 ???????????ChannelFuture f = b.connect(host, port).sync(); ???????????// 等待客户端链路关闭 ???????????f.channel().closeFuture().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(args[0]); ???????????} catch (NumberFormatException e) { ???????????} ???????} ???????new EchoClient().connect(port, "127.0.0.1"); ???}}

package com.zhaowb.netty.ch5_1;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandlerAdapter;import io.netty.channel.ChannelHandlerContext;public class EchoClientHandler extends ChannelHandlerAdapter { ???private int counter; ???static final String ECHO_REQ = "hello.$_"; ???public EchoClientHandler() { ???} ???@Override ???public void channelActive(ChannelHandlerContext ctx) throws Exception { ???????System.out.println("链接成功！"); ???????for (int i = 0; i < 10; i++) { ???????????ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes())); ???????} ???} ???@Override ???public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ???????System.out.println("This is " + ++counter + "times receive server : [" + msg.toString() + "]"); ???} ???@Override ???public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ???????ctx.flush(); ???} ???@Override ???public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ???????cause.printStackTrace(); ???????ctx.close(); ???}}