网络应用，例如服务器和客户端协议。Netty提供了- -种新的方式来使开发网络应用程序，这种新的方式使得它很容易使用和有很强的扩展性。Netty的内部实现是很复杂的，但是Netty提供了简单易用的api从网络处理代码中解耦网络应用程序通常需要有较高的可扩展性，无论是Netty还是其他的基于JavaNIO的框架，都会提供可扩展性的解决方案。Netty中-一个关键组成部分是它的异步特性，本章将讨论同步(阻塞)和异步(非阻塞)的10来说明为什么使用异步代码来解决扩展性问题以及如何使用异步。
架构的组成
NIO模型 同步非阻塞 NIO有同步阻塞和同步非阻塞两种模式，一般讲的是同步非阻塞，服务器实现模式为一个请求一个线程，但客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有I/O请求时才启动一个线程进行处理。 AIO模型 异步非阻塞 服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程进行处理， 注：AIO又称为NIO2.0，在JDK7才开始支持。
为什么Netty使用NIO而不是AIO？
Netty不看重Windows上的使用，在Linux系统上，AIO的底层实现仍使用EPOLL，没有很好实现AIO，因此在性能上没有明显的优势，而且被JDK封装了一层不容易深度优化 Netty整体架构是reactor模型, 而AIO是proactor模型, 混合在一起会非常混乱,把AIO也改造成reactor模型看起来是把epoll绕个弯又绕回来 AIO还有个缺点是接收数据需要预先分配缓存, 而不是NIO那种需要接收时才需要分配缓存, 所以对连接数量非常大但流量小的情况, 内存浪费很多 Linux上AIO不够成熟，处理回调结果速度跟不到处理需求，比如外卖员太少，顾客太多，供不应求，造成处理速度有瓶颈（待验证） 作者原话： Not faster than NIO (epoll) on unix systems (which is true) There is no daragram suppport Unnecessary threading model (too much abstraction without usage)
在学习Netty之前，我们先回顾下FNIO通信步骤: 1创建ServerSocketChannel,.它配置非阻塞模式 2绑定监听，配置TCP参数，录入backlog大小等 3创建-一个独立的I0线程，用于轮询多路复用器Selector 4创建Selector,将之前创建的ServerSocketChannel注册到Selector上，并设置监听标识位 SelectionKey.ACCEPT 5启动I0线程，在循环体中执行Selector.select()方法， 轮询就绪的通道 6当轮询到啦处于就绪的通道时，需要进行判断操作位，如果是ACCEPT状态，说明是新的客户 端接入，则调用accept方法接受新的客户端。 7设置新接入客户端的- -些参数,如非阻塞、并将其通道继续注册到Selector之中，设置监听标 识位等 8如果轮询的通道操作位是READ,则进行读取，构造Buffer对象等 9更细节的还有数据没发送完成继续发送的问题… ok, -一个简单的NIO服务器端程序，就是如此的复制，下面我们看看Netty完成这件事情需要几个步骤。
安装JDK7 下载地址: http://www.oracle.com/te chnetwork/java/javase/archive-139210.html 下载netty包 下载地址: http://netty.io/ Netty实现通信的步骤: 1 创建两个的NIO线程组，-一个专门用于网络事件处理(接受客户端的连接)，另一个 则进行网络通信读写。 存大小 创建一个ServerBootstrap对象，配置Netty的一 系列参数， 例如接受传出数据的缓 3创建一个实际处理数据的类Channellnitializer,进行初始化的准备工作，比如设置 接受传出数据的字符集、格式、已经实际处理数据的接口。 4绑定端口，执行同步阻塞方法等待服务器端启动即可。 如此简单的四个步骤，我们的服务器端就编写完成了，几十行代码就可以把 他完成的健壮、性能稳定，这在我们之前NIO学习的时候是不可能的。 案例：学习网站---- http://ifeve.com/netty5.user.guide/
server.java
package xxx.netty.test; 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 Server { public static void main(String[] args) throws Exception { //NioEventLoopGroup 是用来处理I/O操作的多线程事件循环器 //1 第一个线程组 是用于接收Client端连接的 EventLoopGroup bossGroup = new NioEventLoopGroup(); //2 第二个线程组 是用于实际的业务处理操作的 EventLoopGroup workerGroup = new NioEventLoopGroup(); //3 创建一个辅助类Bootstrap，就是对我们的Server进行一系列的配置 ServerBootstrap b = new ServerBootstrap(); //把俩个工作线程组加入进来 b.group(bossGroup, workerGroup) //我要指定使用NioServerSocketChannel这种类型的通道 .channel(NioServerSocketChannel.class) //一定要使用 childHandler 去绑定具体的 事件处理器 .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(new ServerHandler()); } }); //绑定指定的端口 进行监听 ChannelFuture f = b.bind(8765).sync(); //Thread.sleep(1000000); f.channel().closeFuture().sync(); bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } }ServerHandler.java
package xxx.netty.test; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFuture; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; public class ServerHandler extends ChannelHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //这里的byteBuf被netty封装过了，不用使用filp它自动回复位 //do something msg ByteBuf buf = (ByteBuf)msg; byte[] data = new byte[buf.readableBytes()]; buf.readBytes(data); String request = new String(data, "utf-8"); System.out.println("Server: " + request); //写给客户端 String response = "我是反馈的信息"; ctx.writeAndFlush(Unpooled.copiedBuffer("888".getBytes())); //.addListener(ChannelFutureListener.CLOSE); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }client.java
package xxx.netty.test; import io.netty.bootstrap.Bootstrap; 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.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; public class Client { public static void main(String[] args) throws Exception { EventLoopGroup workgroup = new NioEventLoopGroup(); Bootstrap b = new Bootstrap(); b.group(workgroup) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(new ClientHandler()); } }); ChannelFuture cf1 = b.connect("127.0.0.1", 8765).sync(); //buf cf1.channel().writeAndFlush(Unpooled.copiedBuffer("777".getBytes())); cf1.channel().closeFuture().sync(); workgroup.shutdownGracefully(); } }clientHandler.java
package xxx.netty.test; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; public class ClientHandler extends ChannelHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { //do something msg ByteBuf buf = (ByteBuf)msg; byte[] data = new byte[buf.readableBytes()]; buf.readBytes(data); String request = new String(data, "utf-8"); System.out.println("Client: " + request); } finally { ReferenceCountUtil.release(msg); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }ClientHandler.java
package xxx.netty.test; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; public class ClientHandler extends ChannelHandlerAdapter { @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { //do something msg ByteBuf buf = (ByteBuf)msg; byte[] data = new byte[buf.readableBytes()]; buf.readBytes(data); String request = new String(data, "utf-8"); System.out.println("Client: " + request); } finally { ReferenceCountUtil.release(msg); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }server.java
package xxx.netty.helloworld; 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 Server { public static void main(String[] args) throws Exception { //1 创建线两个程组 //一个是用于处理服务器端接收客户端连接的 //一个是进行网络通信的（网络读写的） EventLoopGroup pGroup = new NioEventLoopGroup(); EventLoopGroup cGroup = new NioEventLoopGroup(); //2 创建辅助工具类，用于服务器通道的一系列配置 ServerBootstrap b = new ServerBootstrap(); b.group(pGroup, cGroup) //绑定俩个线程组 .channel(NioServerSocketChannel.class) //指定NIO的模式 .option(ChannelOption.SO_BACKLOG, 1024) //设置tcp缓冲区 .option(ChannelOption.SO_SNDBUF, 32*1024) //设置发送缓冲大小 .option(ChannelOption.SO_RCVBUF, 32*1024) //这是接收缓冲大小 .option(ChannelOption.SO_KEEPALIVE, true) //保持连接 .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { //3 在这里配置具体数据接收方法的处理 sc.pipeline().addLast(new ServerHandler()); } }); //4 进行绑定 ChannelFuture cf1 = b.bind(8765).sync(); //ChannelFuture cf2 = b.bind(8764).sync();//server端也可以开启多个端口 //5 等待关闭 cf1.channel().closeFuture().sync(); //cf2.channel().closeFuture().sync();释放资源 pGroup.shutdownGracefully(); cGroup.shutdownGracefully(); } }ServerHandler.java
package xxx.netty.helloworld; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelFutureListener; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; public class ServerHandler extends ChannelHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("server channel active... "); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "utf-8"); System.out.println("Server :" + body ); String response = "进行返回给客户端的响应：" + body ; ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes())); //.addListener(ChannelFutureListener.CLOSE); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { System.out.println("读完了"); ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception { ctx.close(); } }client.java
package xxx.netty.helloworld; import io.netty.bootstrap.Bootstrap; 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.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; public class Client { public static void main(String[] args) throws Exception{ EventLoopGroup group = new NioEventLoopGroup(); Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(new ClientHandler()); } }); ChannelFuture cf1 = b.connect("127.0.0.1", 8765).sync(); //ChannelFuture cf2 = b.connect("127.0.0.1", 8764).sync(); //发送消息 Thread.sleep(1000); cf1.channel().writeAndFlush(Unpooled.copiedBuffer("777".getBytes())); cf1.channel().writeAndFlush(Unpooled.copiedBuffer("666".getBytes())); //cf2.channel().writeAndFlush(Unpooled.copiedBuffer("888".getBytes())); Thread.sleep(2000); cf1.channel().writeAndFlush(Unpooled.copiedBuffer("888".getBytes())); //cf2.channel().writeAndFlush(Unpooled.copiedBuffer("666".getBytes())); cf1.channel().closeFuture().sync(); //cf2.channel().closeFuture().sync(); group.shutdownGracefully(); } }ClientHandler.java
package xxx.netty.helloworld; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelPromise; import io.netty.util.ReferenceCountUtil; public class ClientHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { ByteBuf buf = (ByteBuf) msg; byte[] req = new byte[buf.readableBytes()]; buf.readBytes(req); String body = new String(req, "utf-8"); System.out.println("Client :" + body ); String response = "收到服务器端的返回信息：" + body; } finally { ReferenceCountUtil.release(msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }长连接：一个连接一直开着可以多次交互 短连接：一个连接交互完就断开
将netty程序打包jar，单独运行：
在maven项目中打包需要配置的pom文件
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>netty_deploy</groupId> <artifactId>netty_deploy</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>netty_deploy</name> <url>http://maven.apache.org</url> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> </properties> <build> <!-- 配置文件 --> <resources> <resource> <targetPath>${project.build.directory}/classes</targetPath> <directory>src/main/resources</directory> <filtering>true</filtering> <includes> <include>**/*.xml</include> <include>**/*.properties</include> </includes> </resource> <resource> <targetPath>${project.build.directory}/classes/META-INF</targetPath> <directory>src/main/resources</directory> <filtering>true</filtering> <includes> <include>spring-context.xml</include> </includes> </resource> </resources> <pluginManagement> <plugins> <!-- 解决Maven插件在Eclipse内执行了一系列的生命周期引起冲突 --> <plugin> <groupId>org.eclipse.m2e</groupId> <artifactId>lifecycle-mapping</artifactId> <version>1.0.0</version> <configuration> <lifecycleMappingMetadata> <pluginExecutions> <pluginExecution> <pluginExecutionFilter> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-dependency-plugin</artifactId> <versionRange>[2.0,)</versionRange> <goals> <goal>copy-dependencies</goal> </goals> </pluginExecutionFilter> <action> <ignore /> </action> </pluginExecution> </pluginExecutions> </lifecycleMappingMetadata> </configuration> </plugin> </plugins> </pluginManagement> <plugins> <!-- 打包jar文件时，配置manifest文件，加入lib包的jar依赖 --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-jar-plugin</artifactId> <configuration> <classesDirectory>target/classes/</classesDirectory> <archive> <manifest> <mainClass>tpri.mina.execute.Main</mainClass> <!-- 打包时 MANIFEST.MF文件不记录的时间戳版本 --> <useUniqueVersions>false</useUniqueVersions> <addClasspath>true</addClasspath> <classpathPrefix>lib/</classpathPrefix> </manifest> <manifestEntries> <Class-Path>.</Class-Path> </manifestEntries> </archive> </configuration> </plugin> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-dependency-plugin</artifactId> <executions> <execution> <id>copy-dependencies</id> <phase>package</phase> <goals> <goal>copy-dependencies</goal> </goals> <configuration> <type>jar</type> <includeTypes>jar</includeTypes> <useUniqueVersions>false</useUniqueVersions> <outputDirectory> ${project.build.directory}/lib </outputDirectory> </configuration> </execution> </executions> </plugin> </plugins> </build> <dependencies> </dependencies> </project>1.指定main程序类 2.生成lib 3.指定依赖jar在lib下路径（这个maven的pom代码百度）
熟悉tcp编程的可能都知道，无论是服务器端还是客户端，当我们读取或者发送数据的时候，都需要考虑TCP底层的粘包/拆包机制。 TCP是一个“流"协议，所谓流就是没有界限的遗传数据。大家可以想象下如果河里的水就好比数据，他们是连成- -片的，没有分界线，TCP底层并不了解上层的业务数据具体的含义，它会根据TCP缓冲区的实际情况进行包的划分，也就是说，在业务上，我们一个完整的包可能会被TCP分成多个包进行发送，也可能把多个小包封装成-一个大的数据包发送出去，这就是所谓的TCP粘包、拆包问题。分析TCP粘包、拆包问题的产生原因: 1应用程序write写入的字节大小大于套接口发送缓冲区的大小 2进行MSS大小的TCP分段 3以太网帧的payload大于MTU进行IP分片
流数据的传输处理 一个小的Socket Buffer问题 在基于流的传输里比如TCP/IP，接收到的数据会先被存储到一个socket接收缓冲里。不幸的是，基于流的传输并不是一个数据包队列，而是一个字节队列。即使你发送了2个独立的数据包，操作系统也不会作为2个消息处理而仅仅是作为一连串的字节而言。因此这是不能保证你远程写入的数据就会准确地读取。举个例子，让我们假设操作系统的TCP/TP协议栈已经接收了3个数据包：
由于基于流传输的协议的这种普通的性质，在你的应用程序里读取数据的时候会有很高的可能性被分成下面的片段。
因此，一个接收方不管他是客户端还是服务端，都应该把接收到的数据整理成一个或者多个更有意思并且能够让程序的业务逻辑更好理解的数据。在上面的例子中，接收到的数据应该被构造成下面的格式：
再如：
客户端连续发三个：hello netty server端看到的三个连在一起一个字符串，对接受到的数据内容应该是三个分开的hello netty!的字符串，显然这个不是我们想象的结果。
拆包粘包解决方案： 粘包拆包问题的解决方案，根据业界主流协议，的有三种方案: 1消息定长，例如每个报文的大小固定为200个字节，如果不够，空位补空格; 2在包尾部增加特殊字符进行分割，例如加回车等 3讲消息分为消息头和消息体，在消息头中包含表示消息总长度的字段， 然后进行业务逻辑的处理
1分隔符类DelimiterBasedFrameDecoder (自定义分隔符) 2 FixedLengthFrameDecoder (定长)
案例： server.java
package xxx.netty.ende1; import java.nio.ByteBuffer; 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.FixedLengthFrameDecoder; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; public class Server { public static void main(String[] args) throws Exception{ //1 创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的 EventLoopGroup pGroup = new NioEventLoopGroup(); EventLoopGroup cGroup = new NioEventLoopGroup(); //2 创建服务器辅助类 ServerBootstrap b = new ServerBootstrap(); b.group(pGroup, cGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .option(ChannelOption.SO_SNDBUF, 32*1024) .option(ChannelOption.SO_RCVBUF, 32*1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { //设置特殊分隔符 ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes()); sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf)); //设置字符串形式的解码 sc.pipeline().addLast(new StringDecoder()); sc.pipeline().addLast(new ServerHandler()); } }); //4 绑定连接 ChannelFuture cf = b.bind(8765).sync(); //等待服务器监听端口关闭 cf.channel().closeFuture().sync(); pGroup.shutdownGracefully(); cGroup.shutdownGracefully(); } }ServerHandler.java
package xxx.netty.ende1; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; public class ServerHandler extends ChannelHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println(" server channel active... "); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { String request = (String)msg; System.out.println("Server :" + msg); String response = "服务器响应：" + msg + "$_"; ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes())); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception { ctx.close(); } }client.java
package xxx.netty.ende1; 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.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.FixedLengthFrameDecoder; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; public class Client { public static void main(String[] args) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { // ByteBuf buf = Unpooled.copiedBuffer("$_".getBytes()); sc.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, buf)); sc.pipeline().addLast(new StringDecoder()); sc.pipeline().addLast(new ClientHandler()); } }); ChannelFuture cf = b.connect("127.0.0.1", 8765).sync(); cf.channel().writeAndFlush(Unpooled.wrappedBuffer("bbbb$_".getBytes())); cf.channel().writeAndFlush(Unpooled.wrappedBuffer("cccc$_".getBytes())); //等待客户端端口关闭 cf.channel().closeFuture().sync(); group.shutdownGracefully(); } }ClientHandler.java
package xxx.netty.ende1; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; public class ClientHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("client channel active... "); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { String response = (String)msg; System.out.println("Client: " + response); } finally { ReferenceCountUtil.release(msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }Server.java
package xxx.netty.ende2; import java.nio.ByteBuffer; 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.FixedLengthFrameDecoder; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; public class Server { public static void main(String[] args) throws Exception{ //1 创建2个线程，一个是负责接收客户端的连接。一个是负责进行数据传输的 EventLoopGroup pGroup = new NioEventLoopGroup(); EventLoopGroup cGroup = new NioEventLoopGroup(); //2 创建服务器辅助类 ServerBootstrap b = new ServerBootstrap(); b.group(pGroup, cGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .option(ChannelOption.SO_SNDBUF, 32*1024) .option(ChannelOption.SO_RCVBUF, 32*1024) .childHandler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { //设置定长字符串接收 sc.pipeline().addLast(new FixedLengthFrameDecoder(5)); //设置字符串形式的解码 sc.pipeline().addLast(new StringDecoder()); sc.pipeline().addLast(new ServerHandler()); } }); //4 绑定连接 ChannelFuture cf = b.bind(8765).sync(); //等待服务器监听端口关闭 cf.channel().closeFuture().sync(); pGroup.shutdownGracefully(); cGroup.shutdownGracefully(); } }ServerHandler.java
package xxx.netty.ende2; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; public class ServerHandler extends ChannelHandlerAdapter { @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println(" server channel active... "); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { String request = (String)msg; System.out.println("Server :" + msg); String response = request ; ctx.writeAndFlush(Unpooled.copiedBuffer(response.getBytes())); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) throws Exception { } }Client.java
package xxx.netty.ende2; 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.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.FixedLengthFrameDecoder; import io.netty.handler.codec.string.StringDecoder; import io.netty.handler.codec.string.StringEncoder; public class Client { public static void main(String[] args) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(new FixedLengthFrameDecoder(5)); sc.pipeline().addLast(new StringDecoder()); sc.pipeline().addLast(new ClientHandler()); } }); ChannelFuture cf = b.connect("127.0.0.1", 8765).sync(); cf.channel().writeAndFlush(Unpooled.wrappedBuffer("aaaaabbbbb".getBytes())); cf.channel().writeAndFlush(Unpooled.copiedBuffer("ccccccc".getBytes())); //等待客户端端口关闭 cf.channel().closeFuture().sync(); group.shutdownGracefully(); } }ClientHandler.java
package xxx.netty.ende2; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; public class ClientHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { System.out.println("client channel active... "); } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { String response = (String)msg; System.out.println("Client: " + response); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { } }定长虽然可以实现，但是分割符才是最常用的。
编解码技术，说白了就是java序列化技术，序列化目的就两个，第- -进行网络传输，第二对象持久化。虽然我们可以使用java进行对象序列化，netty 去传输，但是java序列化的硬伤太多，比如java序列化没法跨语言、序列化后码流太大、序列化性能太低 等等… 主流的编解码框架:| JBoss的Marshalling包 google的Protobuf 基于Protobuf的Kyro MessagePack框架
我们使用JBoss的Marshalling来解析，使用之前先认识下java解压压缩的工具类 传送文件解压缩的工具类案例： GzipUtils.java
package xxx.utils; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.FileInputStream; import java.io.FileOutputStream; import java.util.zip.GZIPInputStream; import java.util.zip.GZIPOutputStream; public class GzipUtils { public static byte[] gzip(byte[] data) throws Exception{ ByteArrayOutputStream bos = new ByteArrayOutputStream(); GZIPOutputStream gzip = new GZIPOutputStream(bos); gzip.write(data); gzip.finish(); gzip.close(); byte[] ret = bos.toByteArray(); bos.close(); return ret; } public static byte[] ungzip(byte[] data) throws Exception{ ByteArrayInputStream bis = new ByteArrayInputStream(data); GZIPInputStream gzip = new GZIPInputStream(bis); byte[] buf = new byte[1024]; int num = -1; ByteArrayOutputStream bos = new ByteArrayOutputStream(); while((num = gzip.read(buf, 0 , buf.length)) != -1 ){ bos.write(buf, 0, num); } gzip.close(); bis.close(); byte[] ret = bos.toByteArray(); bos.flush(); bos.close(); return ret; } public static void main(String[] args) throws Exception{ //读取文件 String readPath = System.getProperty("user.dir") + File.separatorChar + "sources" + File.separatorChar + "006.jpg"; File file = new File(readPath); FileInputStream in = new FileInputStream(file); byte[] data = new byte[in.available()]; in.read(data); in.close(); System.out.println("文件原始大小:" + data.length); //测试压缩 byte[] ret1 = GzipUtils.gzip(data); System.out.println("压缩之后大小:" + ret1.length); byte[] ret2 = GzipUtils.ungzip(ret1); System.out.println("还原之后大小:" + ret2.length); //写出文件 String writePath = System.getProperty("user.dir") + File.separatorChar + "receive" + File.separatorChar + "006.jpg"; FileOutputStream fos = new FileOutputStream(writePath); fos.write(ret2); fos.close(); } }TestSystemProperty.java
package xxx.utils; public class TestSystemProperty { public static void main(String[] args) { System.out.println("java版本号：" + System.getProperty("java.version")); // java版本号 System.out.println("Java提供商名称：" + System.getProperty("java.vendor")); // Java提供商名称 System.out.println("Java提供商网站：" + System.getProperty("java.vendor.url")); // Java提供商网站 System.out.println("jre目录：" + System.getProperty("java.home")); // Java，哦，应该是jre目录 System.out.println("Java虚拟机规范版本号：" + System.getProperty("java.vm.specification.version")); // Java虚拟机规范版本号 System.out.println("Java虚拟机规范提供商：" + System.getProperty("java.vm.specification.vendor")); // Java虚拟机规范提供商 System.out.println("Java虚拟机规范名称：" + System.getProperty("java.vm.specification.name")); // Java虚拟机规范名称 System.out.println("Java虚拟机版本号：" + System.getProperty("java.vm.version")); // Java虚拟机版本号 System.out.println("Java虚拟机提供商：" + System.getProperty("java.vm.vendor")); // Java虚拟机提供商 System.out.println("Java虚拟机名称：" + System.getProperty("java.vm.name")); // Java虚拟机名称 System.out.println("Java规范版本号：" + System.getProperty("java.specification.version")); // Java规范版本号 System.out.println("Java规范提供商：" + System.getProperty("java.specification.vendor")); // Java规范提供商 System.out.println("Java规范名称：" + System.getProperty("java.specification.name")); // Java规范名称 System.out.println("Java类版本号：" + System.getProperty("java.class.version")); // Java类版本号 System.out.println("Java类路径：" + System.getProperty("java.class.path")); // Java类路径 System.out.println("Java lib路径：" + System.getProperty("java.library.path")); // Java lib路径 System.out.println("Java输入输出临时路径：" + System.getProperty("java.io.tmpdir")); // Java输入输出临时路径 System.out.println("Java编译器：" + System.getProperty("java.compiler")); // Java编译器 System.out.println("Java执行路径：" + System.getProperty("java.ext.dirs")); // Java执行路径 System.out.println("操作系统名称：" + System.getProperty("os.name")); // 操作系统名称 System.out.println("操作系统的架构：" + System.getProperty("os.arch")); // 操作系统的架构 System.out.println("操作系统版本号：" + System.getProperty("os.version")); // 操作系统版本号 System.out.println("文件分隔符：" + System.getProperty("file.separator")); // 文件分隔符 System.out.println("路径分隔符：" + System.getProperty("path.separator")); // 路径分隔符 System.out.println("直线分隔符：" + System.getProperty("line.separator")); // 直线分隔符 System.out.println("操作系统用户名：" + System.getProperty("user.name")); // 用户名 System.out.println("操作系统用户的主目录：" + System.getProperty("user.home")); // 用户的主目录 System.out.println("当前程序所在目录：" + System.getProperty("user.dir")); // 当前程序所在目录 } } java 中 byte[]、File、InputStream 互相转换：将file、fileinputstream转换为byte数组 使用available(): 1、将File、FileInputStream 转换为byte数组： File file = new File("test.txt"); InputStream input = new FileInputStream(file); byte[] byt = new byte[input.available()]; input.read(byt); 2、将byte数组转换为InputStream： byte[] byt = new byte[1024]; InputStream input = new ByteArrayInputStream(byt); 3、将byte数组转换为File： File file = new File(''); OutputStream output = new FileOutputStream(file); BufferedOutputStream bufferedOutput = new BufferedOutputStream(output); bufferedOutput.write(byt);下面是编解码的具体案例： Req.java
package xxx.netty.serial; import java.io.Serializable; public class Req implements Serializable{ private static final long SerialVersionUID = 1L; private String id ; private String name ; private String requestMessage ; private byte[] attachment; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getRequestMessage() { return requestMessage; } public void setRequestMessage(String requestMessage) { this.requestMessage = requestMessage; } public byte[] getAttachment() { return attachment; } public void setAttachment(byte[] attachment) { this.attachment = attachment; } }Resp.java
package xxx.netty.serial; import java.io.Serializable; public class Resp implements Serializable{ private static final long serialVersionUID = 1L; private String id; private String name; private String responseMessage; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getResponseMessage() { return responseMessage; } public void setResponseMessage(String responseMessage) { this.responseMessage = responseMessage; } }Server.java
package bhz.netty.serial; 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; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; public class Server { public static void main(String[] args) throws Exception{ EventLoopGroup pGroup = new NioEventLoopGroup(); EventLoopGroup cGroup = new NioEventLoopGroup(); ServerBootstrap b = new ServerBootstrap(); b.group(pGroup, cGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) //设置日志 .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); sc.pipeline().addLast(new ServerHandler()); } }); ChannelFuture cf = b.bind(8765).sync(); cf.channel().closeFuture().sync(); pGroup.shutdownGracefully(); cGroup.shutdownGracefully(); } }ServerHandler.java
package xxx.netty.serial; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import java.io.File; import java.io.FileOutputStream; import bhz.utils.GzipUtils; public class ServerHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { Req req = (Req)msg; System.out.println("Server : " + req.getId() + ", " + req.getName() + ", " + req.getRequestMessage()); byte[] attachment = GzipUtils.ungzip(req.getAttachment()); String path = System.getProperty("user.dir") + File.separatorChar + "receive" + File.separatorChar + "001.jpg"; FileOutputStream fos = new FileOutputStream(path); fos.write(attachment); fos.close(); Resp resp = new Resp(); resp.setId(req.getId()); resp.setName("resp" + req.getId()); resp.setResponseMessage("响应内容" + req.getId()); ctx.writeAndFlush(resp);//.addListener(ChannelFutureListener.CLOSE); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }Client.java
package xxx.netty.serial; import io.netty.bootstrap.Bootstrap; 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.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import java.io.File; import java.io.FileInputStream; import bhz.utils.GzipUtils; public class Client { public static void main(String[] args) throws Exception{ EventLoopGroup group = new NioEventLoopGroup(); Bootstrap b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); sc.pipeline().addLast(new ClientHandler()); } }); ChannelFuture cf = b.connect("127.0.0.1", 8765).sync(); for(int i = 0; i < 5; i++ ){ Req req = new Req(); req.setId("" + i); req.setName("pro" + i); req.setRequestMessage("数据信息" + i); String path = System.getProperty("user.dir") + File.separatorChar + "sources" + File.separatorChar + "001.jpg"; File file = new File(path); FileInputStream in = new FileInputStream(file); byte[] data = new byte[in.available()]; in.read(data); in.close(); req.setAttachment(GzipUtils.gzip(data)); cf.channel().writeAndFlush(req); } cf.channel().closeFuture().sync(); group.shutdownGracefully(); } }ClientHandler.java
package xxx.netty.serial; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; import io.netty.util.ReferenceCountUtil; public class ClientHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { try { Resp resp = (Resp)msg; System.out.println("Client : " + resp.getId() + ", " + resp.getName() + ", " + resp.getResponseMessage()); } finally { ReferenceCountUtil.release(msg); } } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }MarshallingCodeCFactory.java
package xxx.netty.serial; import io.netty.handler.codec.marshalling.DefaultMarshallerProvider; import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider; import io.netty.handler.codec.marshalling.MarshallerProvider; import io.netty.handler.codec.marshalling.MarshallingDecoder; import io.netty.handler.codec.marshalling.MarshallingEncoder; import io.netty.handler.codec.marshalling.UnmarshallerProvider; import org.jboss.marshalling.MarshallerFactory; import org.jboss.marshalling.Marshalling; import org.jboss.marshalling.MarshallingConfiguration; /** * Marshalling工厂 * @author（alienware） * @since 2014-12-16 */ public final class MarshallingCodeCFactory { /** * 创建Jboss Marshalling解码器MarshallingDecoder * @return MarshallingDecoder */ public static MarshallingDecoder buildMarshallingDecoder() { //首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。 final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); //创建了MarshallingConfiguration对象，配置了版本号为5 final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(5); //根据marshallerFactory和configuration创建provider UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration); //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度 MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024 * 1024 * 1); return decoder; } /** * 创建Jboss Marshalling编码器MarshallingEncoder * @return MarshallingEncoder */ public static MarshallingEncoder buildMarshallingEncoder() { final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(5); MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration); //构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组 MarshallingEncoder encoder = new MarshallingEncoder(provider); return encoder; } }（自己看实际工作很少使用） server.java
package xxx.netty.udp; import io.netty.bootstrap.Bootstrap; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.nio.NioDatagramChannel; public class Server { public void run(int port) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group).channel(NioDatagramChannel.class) .option(ChannelOption.SO_BROADCAST, true) .handler(new ServerHandler()); b.bind(port).sync().channel().closeFuture().await(); } finally { group.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new Server().run(8765); new Server().run(8764); } }ServerHandler.java
package xxx.netty.udp; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.socket.DatagramPacket; import io.netty.util.CharsetUtil; import io.netty.util.internal.ThreadLocalRandom; public class ServerHandler extends SimpleChannelInboundHandler<DatagramPacket> { // 谚语列表 private static final String[] DICTIONARY = { "只要功夫深，铁棒磨成针。", "旧时王谢堂前燕，飞入寻常百姓家。", "洛阳亲友如相问，一片冰心在玉壶。", "一寸光阴一寸金，寸金难买寸光阴。", "老骥伏枥，志在千里。烈士暮年，壮心不已!" }; private String nextQuote() { int quoteId = ThreadLocalRandom.current().nextInt(DICTIONARY.length); return DICTIONARY[quoteId]; } @Override public void messageReceived(ChannelHandlerContext ctx, DatagramPacket packet) throws Exception { String req = packet.content().toString(CharsetUtil.UTF_8); System.out.println(req); if ("谚语字典查询?".equals(req)) { ctx.writeAndFlush( new DatagramPacket(Unpooled.copiedBuffer("谚语查询结果: " + nextQuote(), CharsetUtil.UTF_8), packet.sender())); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); cause.printStackTrace(); } }cilent.java
package xxx.netty.udp; import io.netty.bootstrap.Bootstrap; import io.netty.buffer.Unpooled; import io.netty.channel.Channel; import io.netty.channel.ChannelOption; import io.netty.channel.EventLoopGroup; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.DatagramPacket; import io.netty.channel.socket.nio.NioDatagramChannel; import io.netty.util.CharsetUtil; import java.net.InetSocketAddress; public class Client { public void run(int port) throws Exception { EventLoopGroup group = new NioEventLoopGroup(); try { Bootstrap b = new Bootstrap(); b.group(group).channel(NioDatagramChannel.class) .option(ChannelOption.SO_BROADCAST, true) .handler(new ClientHandler()); Channel ch = b.bind(0).sync().channel(); // 向网段内的所有机器广播UDP消息 ch.writeAndFlush(new DatagramPacket(Unpooled.copiedBuffer("谚语字典查询?", CharsetUtil.UTF_8), new InetSocketAddress("255.255.255.255", port))).sync(); if (!ch.closeFuture().await(15000)) { System.out.println("查询超时!"); } } finally { group.shutdownGracefully(); } } public static void main(String[] args) throws Exception { new Client().run(8765); } }ClientHandler.java
package xxx.netty.udp; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.SimpleChannelInboundHandler; import io.netty.channel.socket.DatagramPacket; import io.netty.util.CharsetUtil; public class ClientHandler extends SimpleChannelInboundHandler<DatagramPacket> { @Override public void messageReceived(ChannelHandlerContext ctx, DatagramPacket msg) throws Exception { String response = msg.content().toString(CharsetUtil.UTF_8); if (response.startsWith("谚语查询结果: ")) { System.out.println(response); ctx.close(); } } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { cause.printStackTrace(); ctx.close(); } }我们需要了解下在真正项目应用中如何去考虐Netty的使用，大体上对于-些参数设置都是根据 服务器性能决定的。这个不是最主要的。 我们需要考虑的问题是两台机器(甚至多台)使用Netty的怎样进行通信，我个人大体上把他分 为三种: 1第一种，使用长连接通道不断开的形式进行通信，也就是服务器和客户端的通道一直处于 开启状态，如果服务器性能足够好，并且我们的客户端数量也比较少的情况下，我还是推荐这种 方式的. 2第二种，一次性批量提交数据，采用短连接方式。也就是我们会把数据保存在本地临时缓 冲区或者临时表里，当达到临界值时进行一次性批量提交， 又或者根据定时任务轮询提交，这种 情况弊端是做不到实时性传输，在对实时性不高的应用程序中可以推荐使用. 3第三种，我们可以使用一-种特殊的长连接，在指定某- -时间之内，服务器与某台客户端没 有任何通信，则断开连接。下次连接则是客户端向服务器发送请求的时候，再次建立连接。但是 这种模式我们需要考虑2个因素: 1如何在超时(即服务器和客户端没有任何通信)后关闭通道?关闭通道后我们又如何 再次建立连接? 2客户端宕机时，我们无需考虑，下次客户端重启之后我们就可以与服务器建立连接， 但是服务器宕机时，我们的客户端如何与服务器进行连接呢?
server.java
package xxx.netty.runtime; 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; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.ReadTimeoutHandler; public class Server { public static void main(String[] args) throws Exception{ EventLoopGroup pGroup = new NioEventLoopGroup(); EventLoopGroup cGroup = new NioEventLoopGroup(); ServerBootstrap b = new ServerBootstrap(); b.group(pGroup, cGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) //设置日志 .handler(new LoggingHandler(LogLevel.INFO)) .childHandler(new ChannelInitializer<SocketChannel>() { protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); sc.pipeline().addLast(new ReadTimeoutHandler(5)); sc.pipeline().addLast(new ServerHandler()); } }); ChannelFuture cf = b.bind(8765).sync(); cf.channel().closeFuture().sync(); pGroup.shutdownGracefully(); cGroup.shutdownGracefully(); } }ServerHandler.java
package xxx.netty.runtime; import io.netty.channel.ChannelHandlerAdapter; import io.netty.channel.ChannelHandlerContext; public class ServerHandler extends ChannelHandlerAdapter{ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { } @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { Request request = (Request)msg; System.out.println("Server : " + request.getId() + ", " + request.getName() + ", " + request.getRequestMessage()); Response response = new Response(); response.setId(request.getId()); response.setName("response" + request.getId()); response.setResponseMessage("响应内容" + request.getId()); ctx.writeAndFlush(response);//.addListener(ChannelFutureListener.CLOSE); } @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.close(); } }client.java
package bhz.netty.runtime; import io.netty.bootstrap.Bootstrap; 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.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.handler.timeout.ReadTimeoutHandler; import java.util.concurrent.TimeUnit; /** * Best Do It */ public class Client { private static class SingletonHolder { static final Client instance = new Client(); } public static Client getInstance(){ return SingletonHolder.instance; } private EventLoopGroup group; private Bootstrap b; private ChannelFuture cf ; private Client(){ group = new NioEventLoopGroup(); b = new Bootstrap(); b.group(group) .channel(NioSocketChannel.class) .handler(new LoggingHandler(LogLevel.INFO)) .handler(new ChannelInitializer<SocketChannel>() { @Override protected void initChannel(SocketChannel sc) throws Exception { sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder()); sc.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder()); //超时handler（当服务器端与客户端在指定时间以上没有任何进行通信，则会关闭响应的通道，主要为减小服务端资源占用） sc.pipeline().addLast(new ReadTimeoutHandler(5)); sc.pipeline().addLast(new ClientHandler()); } }); } public void connect(){ try { this.cf = b.connect("127.0.0.1", 8765).sync(); System.out.println("远程服务器已经连接, 可以进行数据交换.."); } catch (Exception e) { e.printStackTrace(); } } public ChannelFuture getChannelFuture(){ if(this.cf == null){ this.connect(); } if(!this.cf.channel().isActive()){ this.connect(); } return this.cf; } public static void main(String[] args) throws Exception{ final Client c = Client.getInstance(); //c.connect(); ChannelFuture cf = c.getChannelFuture(); for(int i = 1; i <= 3; i++ ){ Request request = new Request(); request.setId("" + i); request.setName("pro" + i); request.setRequestMessage("数据信息" + i); cf.channel().writeAndFlush(request); TimeUnit.SECONDS.sleep(4); } cf.channel().closeFuture().sync(); new Thread(new Runnable() { @Override public void run() { try { System.out.println("进入子线程..."); ChannelFuture cf = c.getChannelFuture(); System.out.println(cf.channel().isActive()); System.out.println(cf.channel().isOpen()); //再次发送数据 Request request = new Request(); request.setId("" + 4); request.setName("pro" + 4); request.setRequestMessage("数据信息" + 4); cf.channel().writeAndFlush(request); cf.channel().closeFuture().sync(); System.out.println("子线程结束."); } catch (InterruptedException e) { e.printStackTrace(); } } }).start(); System.out.println("断开连接,主线程结束.."); } }cilentHandler.java
MarshallingCodeCFactory.java
package xxx.netty.runtime; import io.netty.handler.codec.marshalling.DefaultMarshallerProvider; import io.netty.handler.codec.marshalling.DefaultUnmarshallerProvider; import io.netty.handler.codec.marshalling.MarshallerProvider; import io.netty.handler.codec.marshalling.MarshallingDecoder; import io.netty.handler.codec.marshalling.MarshallingEncoder; import io.netty.handler.codec.marshalling.UnmarshallerProvider; import org.jboss.marshalling.MarshallerFactory; import org.jboss.marshalling.Marshalling; import org.jboss.marshalling.MarshallingConfiguration; /** * Marshalling工厂 * @author（alienware） * @since 2014-12-16 */ public final class MarshallingCodeCFactory { /** * 创建Jboss Marshalling解码器MarshallingDecoder * @return MarshallingDecoder */ public static MarshallingDecoder buildMarshallingDecoder() { //首先通过Marshalling工具类的精通方法获取Marshalling实例对象 参数serial标识创建的是java序列化工厂对象。 final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); //创建了MarshallingConfiguration对象，配置了版本号为5 final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(5); //根据marshallerFactory和configuration创建provider UnmarshallerProvider provider = new DefaultUnmarshallerProvider(marshallerFactory, configuration); //构建Netty的MarshallingDecoder对象，俩个参数分别为provider和单个消息序列化后的最大长度 MarshallingDecoder decoder = new MarshallingDecoder(provider, 1024); return decoder; } /** * 创建Jboss Marshalling编码器MarshallingEncoder * @return MarshallingEncoder */ public static MarshallingEncoder buildMarshallingEncoder() { final MarshallerFactory marshallerFactory = Marshalling.getProvidedMarshallerFactory("serial"); final MarshallingConfiguration configuration = new MarshallingConfiguration(); configuration.setVersion(5); MarshallerProvider provider = new DefaultMarshallerProvider(marshallerFactory, configuration); //构建Netty的MarshallingEncoder对象，MarshallingEncoder用于实现序列化接口的POJO对象序列化为二进制数组 MarshallingEncoder encoder = new MarshallingEncoder(provider); return encoder; } }request.java
package xxx.netty.runtime; import java.io.Serializable; public class Request implements Serializable{ private static final long SerialVersionUID = 1L; private String id ; private String name ; private String requestMessage ; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getRequestMessage() { return requestMessage; } public void setRequestMessage(String requestMessage) { this.requestMessage = requestMessage; } }response.java
package xxx.netty.runtime; import java.io.Serializable; public class Response implements Serializable{ private static final long serialVersionUID = 1L; private String id; private String name; private String responseMessage; public String getId() { return id; } public void setId(String id) { this.id = id; } public String getName() { return name; } public void setName(String name) { this.name = name; } public String getResponseMessage() { return responseMessage; } public void setResponseMessage(String responseMessage) { this.responseMessage = responseMessage; } }