Netty实践

客官°小女子只卖身不卖艺 2022-06-04 09:49 211阅读 0赞

“Netty是一个异步的、基于事件驱动的网络应用框架，用来快速开发高性能Servers、Clients”，这是官网对Netty的定义；以本人对Netty的使用感受而言，可以认为Netty是：

1）对JAVA NIO的高度封装，开发者不需要关注NIO的底层API、ByteBuffer，Netty几乎对所有的NIO API进行了二次封装，包括Channel、ByteBuffer、Selector等，开发者将从以前繁琐而已易于出错的NIO开发中解脱出来。“Netty框架”的API使用简单，“约定俗成”，最终使得开发者写出的代码样式非常相似，易于阅读和扩展。

2）不需要关注网络层（Socket，以及协议TCP有关）的处理细节和异常，Netty已经为我们做好了“开关”。

3）不需要过度关注应用级别的数据序列化机制，Netty支持多种序列化框架，或者说Netty可以作为其他RPC服务的底层通道，比如Protobuf、thrift、avro等。而且Netty提供了易于扩展的多种“ChannelHandler”，开发者可以自定义“编解码”处理器，而不需要像NIO那样“沉溺”于痛苦的ByteBuffer处理中。（你应该想起“半包”处理、附件传递，还有那些让人难忘的slice()、rewind()方法）

4）不需要过度关注所谓的性能和调优，Netty提供了线程池模式（提高并发能力）以及Buffer的重用机制（减少内存Copy），开发者不需要构建、维护复杂的多线程模型和操作队列等（reactor模式），只需要简单的参数配置，即可达成高性能通讯的性能要求；这些，Netty已经准备好了。

Netty其实就是一个NIO框架，它适用于服务器通讯相关的多种应用场景，主要还是针对于TCP协议下，面向Clients端的高并发应用，或者Peer-to-Peer场景下的大量数据持续传输的应用。接下来，我们从实用是角度来了解一下Netty。

**一、入门实例**

NIO，即Non Blocking IO，非阻塞IO，在JAVA中NIO的核心就是Selector机制。简单而言，创建一个Socket Channel，并将其注册到一个Selector上（多路复用器），这个Selector将会“关注”Channel上发生的IO读写事件，并在事件发生（数据就绪）后执行相关的处理逻辑。对于阻塞IO，它需要在read()、write()操作上阻塞而直到数据操作完毕，但是NIO则不需要，只有当Selector检测到此Channel上有事件时才会触发调用read、write操作。

但是NIO并不是严格意义上的“异步IO”（Asynchronous），最大的原因就是Selector本身是阻塞的！！即selector需要通过线程阻塞的方式（其select方法）获取底层通道的事件变更，然后获取SelectionKey列表；那么对于“异步IO”（概念同JDK 7的AIO）在整个操作链路上均不需要任何阻塞（完全基于OS的IO事件），依赖基于事件驱动的Handler做数据处理。目前Netty尚没有集成AIO的相关特性，即Netty本身为非阻塞IO框架。

学习Netty最好的办法，就是下载netty-example.jar的源码文件，逐个学习它的例子即可。

如下为一个简单的Netty实例，本实例主要描述了Client向Server端发送一个String，然后Server将String进行MD5加密，然后再响应给Client；本实例主要演示Client与Server如果使用Netty实现一个“交互式”的请求（request-response），以及如何对数据进行编解码的。

**1、pom.xml**

Java代码   ![收藏代码][icon_star.png]

1.  <dependency>  
2.    <groupId>io.netty</groupId>  
3.    <artifactId>netty-all</artifactId>  
4.    <version>4.0.27.Final</version>  
5.  </dependency>

本实例中，还会用到json-lib,apache commons相关依赖，具体版本请开发者自己选定。

**2、NettyServer.java：**主要为初始化Server Socket，并负责调度相关的IO操作。

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** NettyServer \{  
2.    
3.      **private** **static** **final** Charset UTF\_8 = Charset.forName("utf-8");  
4.    
5.      **private** ChannelFuture future;  
6.    
7.      **private** **boolean** isClosed = **false**;  
8.    
9.      **private** **boolean** init = **false**;  
10.   
11.     **private** ServerBootstrap bootstrap;  
12.   
13.     **public** **void** start() \{  
14.   
15.         **if**(init) \{  
16.             **throw** **new** RuntimeException("Client is already started!");  
17.         \}  
18.         //thread model：one selector thread，and one worker thread pool。  
19.         EventLoopGroup bossGroup = **new** NioEventLoopGroup(1);//more than 1 is not needed!  
20.         EventLoopGroup workerGroup = **new** NioEventLoopGroup(Runtime.getRuntime().availableProcessors() - 1);  
21.         **try** \{  
22.             bootstrap = **new** ServerBootstrap();//create ServerSocket transport。  
23.             bootstrap.group(bossGroup,workerGroup)  
24.                     .channel(NioServerSocketChannel.**class**)  
25.                     .childHandler(**new** ChannelInitializer<SocketChannel>() \{  
26.                         @Override  
27.                         **protected** **void** initChannel(SocketChannel ch) **throws** Exception \{  
28.                             ch.pipeline().addLast(**new** LengthFieldBasedFrameDecoder(10240, 0, 2, 0, 2))  
29.                                     .addLast(**new** StringDecoder(UTF\_8))  
30.                                     .addLast(**new** LengthFieldPrepender(2))  
31.                                     .addLast(**new** StringEncoder(UTF\_8))  
32.                                     .addLast(**new** ServerHandler());  
33.                         \}  
34.                     \}).childOption(ChannelOption.TCP\_NODELAY, **true**);  
35.             future = bootstrap.bind(18080).sync();  
36.             init = **true**;  
37.             //  
38.             System.out.println("server started");  
39.         \} **catch** (Exception e) \{  
40.             isClosed = **true**;  
41.         \} **finally** \{  
42.             **if**(isClosed) \{  
43.                 workerGroup.shutdownGracefully();  
44.                 bossGroup.shutdownGracefully();  
45.                 System.out.println("server closed");  
46.             \}  
47.         \}  
48.     \}  
49.   
50.   
51.     **public** **void** close() \{  
52.         **if**(isClosed) \{  
53.             **return**;  
54.         \}  
55.         **try** \{  
56.             future.channel().close();  
57.         \} **finally** \{  
58.             bootstrap.childGroup().shutdownGracefully();  
59.             bootstrap.group().shutdownGracefully();  
60.         \}  
61.         isClosed = **true**;  
62.         System.out.println("server closed");  
63.     \}  
64.   
65. \}

NettyServer的start方法，创建一个“ServerSocket”绑定在18080端口，内部将启动线程池和Selector，等待Client建立连接。close方法，就是关闭Server。代码中，还有很多Netty有关的API，我们稍后会逐个详解。

**3、ServerHandler.java：**在NettyServer中，我们在pipeline注册了一个ServerHandler，这个handler用来处理Client端实际发送的数据，它将和其他Decoder一起协同，对Socket中read的字节数据进行解码。

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** ServerHandler **extends** SimpleChannelInboundHandler<String> \{  
2.    
3.      @Override  
4.      **public** **void** channelRead0(ChannelHandlerContext ctx, String message) **throws** Exception \{  
5.          System.out.println("from client:" + message);  
6.          JSONObject json = JSONObject.fromObject(message);  
7.          String source = json.getString("source");  
8.    
9.          String md5 = DigestUtils.md5Hex(source);  
10.         //解析成JSON  
11.         json.put("md5Hex",md5);  
12.         ctx.writeAndFlush(json.toString());//write bytes to socket,and flush(clear) the buffer cache.  
13.     \}  
14.   
15.     @Override  
16.     **public** **void** channelReadComplete(ChannelHandlerContext ctx) \{  
17.         ctx.flush();  
18.     \}  
19.   
20.     @Override  
21.     **public** **void** exceptionCaught(ChannelHandlerContext ctx, Throwable cause) \{  
22.         cause.printStackTrace();  
23.         ctx.close();  
24.     \}  
25. \}

如果你对Netty尚没有接触过，那么我在此处简单说一下：

1）StringEncoder：编码器，将string编码为字节数组。

2）LengthFieldPrepender：编码器，将1）编码后的字节数组的前面，prepender一个field（字段站位），用来表示原始字节数组的长度。这个编码器的构造函数需要一个int值，表示lengthField占用几个字节。比如字节数据的长度为120，lengthField配置为2个字节，那么此编码器将会首先写入一个数字120，占用2个字节，然后再写入原始的那120个字节数据。

3）LengthFieldBasedFrameDecoder：解码器，和LengthFieldPrepender对应，首选读取一个2个字节（通过构造函数指定）表示LengthField，比如为120，那么此Decoder将会在此后连续读取120个字节即可。并将这120个字节数组传递到下一个Decoder中。这种手段，称为“字节码成帧”，是socket通讯中比较常用的。

4）StringDecoder：和StringEncoder对应，将指定的字节数组，编码成string字符串。

5）ServerHandler：这个就是我们自定义的Decoder了，它继承了ChannelInboundHandler,它的channelRead0方法获取到的对象，就是上述Decoder解码的结果。

大家还会发现，这些Decoder、Encoder是通过一个addLast方法添加到pipeline中的。这个pipeline内部为一个链表结果，将这些编码、解码器按照顺序排列。对于socket write操作，将会从Tail--》Head依次执行pipeline中的每个Encoder，对于read操作，将会从Head--》Tail依次执行Decoder。这些细节，我们稍后会详细介绍。

那么此实例即可得出：socket channel中read操作得到的字节流，将会依次经过：

1）LengthFieldBasedFrameDecoder：它读取一个lengthField，然后依次读取length个字节，将字节数组放入ByteBuffer，然后传递给下一个Decoder。

2）StringDecoder：将ByteBuffer，根据UTF-8编码成字符串。

3）ServerHandler：接收到字符串message，执行业务逻辑。然后将结果字符串写入channel。

4）StringEncoder：将ServerHandler写入的字符串，使用UTF-8编码为字节数组。

5）LengthFieldPrepender：将字节数组的前面添加一个LengthField，表示字节数组的长度。

6）经过底层Socket将最终的字节数据写入通道并发送。

**4、NettyClient.java：**初始化Socket Channel，和Server建立连接。

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** NettyClient \{  
2.    
3.      **private** **static** **final** Charset UTF\_8 = Charset.forName("utf-8");  
4.    
5.      **private** ClientHandler clientHandler = **new** ClientHandler();  
6.    
7.      **private** Bootstrap bootstrap;  
8.    
9.      **private** ChannelFuture future;  
10.   
11.     **private** **boolean** init = **false**;  
12.   
13.     **private** **boolean** isClosed = **false**;  
14.   
15.     **public** **void** start() \{  
16.         **if**(init) \{  
17.             **throw** **new** RuntimeException("client is already started");  
18.         \}  
19.         //thread model: one worker thread pool,contains selector thread and workers‘.  
20.         EventLoopGroup workerGroup = **new** NioEventLoopGroup(2);//1 is OK  
21.         **try** \{  
22.             bootstrap = **new** Bootstrap();  
23.             bootstrap.group(workerGroup)  
24.                     .channel(NioSocketChannel.**class**) //create SocketChannel transport  
25.                     .option(ChannelOption.CONNECT\_TIMEOUT\_MILLIS,10000)  
26.                     .handler(**new** ChannelInitializer<SocketChannel>() \{  
27.                         @Override  
28.                         **protected** **void** initChannel(SocketChannel ch) **throws** Exception \{  
29.                             ch.pipeline().addLast(**new** LengthFieldBasedFrameDecoder(10240, 0, 2, 0, 2))  
30.                                     .addLast(**new** StringDecoder(UTF\_8))  
31.                                     .addLast(**new** LengthFieldPrepender(2))  
32.                                     .addLast(**new** StringEncoder(UTF\_8))  
33.                                     .addLast(clientHandler);//the same as ServerBootstrap  
34.                         \}  
35.                     \});  
36.             //keep the connection with server，and blocking until closed!   
37.             future = bootstrap.connect(**new** InetSocketAddress("127.0.0.1", 18080)).sync();  
38.             init = **true**;  
39.         \} **catch** (Exception e) \{  
40.             isClosed = **true**;  
41.         \} **finally** \{  
42.             **if**(isClosed) \{  
43.                 workerGroup.shutdownGracefully();  
44.             \}  
45.         \}  
46.     \}  
47.   
48.     **public** **void** close() \{  
49.         **if**(isClosed) \{  
50.             **return**;  
51.         \}  
52.         **try** \{  
53.             future.channel().close();  
54.         \} **finally** \{  
55.             bootstrap.group().shutdownGracefully();  
56.         \}  
57.         isClosed = **true**;  
58.     \}  
59.   
60.     /\*\* 
61.      \* 发送消息 
62.      \* @param message 
63.      \* @return 
64.      \* @throws Exception 
65.      \*/  
66.     **public** String send(String message) **throws** Exception \{  
67.         **if**(isClosed || !init) \{  
68.             **throw** **new** RuntimeException("client has been closed!");  
69.         \}  
70.         //send a request call,and blocking until recevie a response from server.   
71.         **return** clientHandler.call(message,future.channel());  
72.     \}  
73.       
74. \}

我们可以看到NettyClient中有一个send方法，这个方法就是用来向Server发送消息的。

**5、ClientHandler.java：**配合NettyClient，完成消息的读取操作以及业务处理。

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** ClientHandler **extends** SimpleChannelInboundHandler<String> \{  
2.    
3.      //key is sequence ID，value is response message.  
4.      **private** Map<Integer,String> response = **new** ConcurrentHashMap<Integer, String>();  
5.    
6.      //key is sequence ID，value is request thread.  
7.      **private** **final** Map<Integer,Thread> waiters = **new** ConcurrentHashMap<Integer, Thread>();  
8.    
9.      **private** **final** AtomicInteger sequence = **new** AtomicInteger();  
10.   
11.   
12.     @Override  
13.     **public** **void** channelActive(ChannelHandlerContext ctx) **throws** Exception \{  
14.         //当channel就绪后。  
15.         System.out.println("client channel is ready!");  
16.         //ctx.writeAndFlush("started");//阻塞知道发送完毕  
17.     \}  
18.   
19.     @Override  
20.     **public** **void** channelRead0(ChannelHandlerContext ctx, String message) **throws** Exception \{  
21.         JSONObject json = JSONObject.fromObject(message);  
22.         Integer id = json.getInt("id");  
23.         response.put(id,json.getString("md5Hex"));  
24.   
25.         Thread thread = waiters.remove(id);//读取到response后，从waiters中移除并唤醒线程。  
26.         **synchronized** (thread) \{  
27.             thread.notifyAll();  
28.         \}  
29.     \}  
30.   
31.   
32.     **public** String call(String message,Channel channel) **throws** Exception \{  
33.         **int** id = sequence.incrementAndGet();//产生一个ID，并与当前request绑定  
34.         Thread current = Thread.currentThread();  
35.         waiters.put(id,current);  
36.         JSONObject json = **new** JSONObject();  
37.         json.put("id",id);  
38.         json.put("source",message);  
39.         channel.writeAndFlush(json.toString());  
40.         **while** (!response.containsKey(id)) \{  
41.             **synchronized** (current) \{  
42.                 current.wait();//阻塞请求调用者线程，直到收到响应响应  
43.             \}  
44.         \}  
45.         waiters.remove(id);  
46.         **return** response.remove(id);  
47.   
48.     \}  
49.   
50. \}

ClientHandler稍微有些复杂，除了call方法之外，其他的两个方法都很好理解。channelActive方法，此处没有实际用途，仅作展示，当channel准备待续后（已经在selector上注册READ/WRITE事件）将会执行这个方法，通常我们可以在此方法中执行首次的write操作，比如Client在连接建立成功后，首先向Server端发送一个消息。

call方法是我们自定义的方法，这个方法由NettClient.send方法直接调用。我们最终表达的目的就是让NettyClient在调用send方法时，只有当数据发送到Server端，且sever端响应的数据返回后才能让send方法接触阻塞，就像模拟一次RPC请求。因为在并发调用时，socket本身无法分拣数据是由那个线程发送，那么我们就为每个send操作首先创建一个sequence ID，server响应数据也会把此ID反馈回来；当数据通过channel发送后，当前线程阻塞。那么channelRead方法接收到响应后，将结果添加到response map中，并唤醒此ID对应的线程，那么此时call方法再继续执行并从response中获取结果。

上述代码，只是演示了一个Client与serve端的交互模型，一个RPC模型，代码如果在production环境中使用，还需要更加细致的调整。

**6、Test：**测试类。

Java代码   ![收藏代码][icon_star.png]

1.  **public** **class** NettyTestMain \{  
2.    
3.      **public** **static** **void** main(String\[\] args) **throws** Exception \{  
4.          NettyServer nettyServer = **new** NettyServer();  
5.          nettyServer.start();//启动server  
6.          Thread.sleep(3000);  
7.    
8.          NettyClient nettyClient = **new** NettyClient();  
9.          nettyClient.start();  
10.   
11.         **try** \{  
12.             **for** (**int** i = 0; i < 5; i++) \{  
13.                 String response = nettyClient.send(RandomStringUtils.random(32, **true**, **true**));  
14.                 System.out.println("response:" + response);  
15.             \}  
16.         \} **catch** (Exception e) \{  
17.             e.printStackTrace();  
18.         \} **finally** \{  
19.             nettyClient.close();  
20.         \}  
21.   
22.         nettyServer.close();  
23.   
24.     \}  
25. \}

如果一切顺利，那么此NettyTestMain将会正常打印结果。

我们通过上述的例子，已经认识到了Netty中几个核心的API：EventLoopGroup、ServerBootstrap/Bootstrap、ChannelHandler、ChannelFuture、Channel，以及ByteBuf、Listener等。接下来我们将会详细介绍每个API的设计原理以及使用规范。

我们在使用Netty的时候，并不会直接接触到NIO的API，不过Netty本身还是NIO的封装，我们会在合适的地方讲解上述API与JAVA NIO的封装关系。

**二、Channel接口：**通道，其实NIO中有个同名的“Channel”，不过这两个接口，本身没有任何“依赖关系”，但是它们表达的语义是一致的：一个逻辑通道，维护socket上IO的write、read、connect等相关操作。Netty并没有通过“继承”的方式改变NIO Channel的行为，而是重新设计了一个Channel接口，这主要原因就是1）JAVA NIO中Channel的继承关系非常复杂，如果继承Channel来实现框架，那么需要调整的API（子接口、子类）将会“众多”而无法实施，Netty几乎无法通过简单的方式达成自己的设计目的 2）JAVA NIO中Channel实现更加底层而且固化，Netty的初衷是一个NIO框架，那么它所能做的“封装”、“组合”、“统一操作视图”，所以Netty只需要通过自己的Channel表达“IO语义”，然后内部组合NIO Channel的实例即可。

Channel内部持有一个javaChannel()的方法，这个方法将会返回一个SelectableChannel实例，即NIO Channel的实例，对于Server而言为ServerSocketChannel，对于Client而言为SocketChannel。在上述例子中我们可以看到Bootstrap、ServerBootstrap初始化之后都需要在channel(Channel channel)方法中指定内部需要创建的channel类型。

Channel中提供了大量的Socket操作方法，比如connect、bind、close、write、read；这是NIO Channel接口中所没有的，而是NIO SocketChannel/ServerSocketChannel中才有的行为，从这一点可以看出Channel其实是为了“统一视图”（对于Netty而言，Channel接口可描述为Server或Client端）。而且还需要声明，Channel中所有的操作均是异步的，IO操作都会返回一个ChannelFuture实例：

*  ChannelFuture connect(SocketAddress remoteAddress);
 *  ChannelFuture close()：关闭底层Socket，并取消SelectionKey的注册。此处将会唤醒那些阻塞在Channel上的其他操作，比如CloseFuturue.sync()方法。一旦Channel关闭，Channel以及相关的Future将不可重用。此方法内部将会依次执行pipeline中ChannelOutboundHandler的close()方法。
 *  ChannelFuture write(Object msg)：将当前msg写入Channel，但并不是通过Socket发送出去；此方法只是将msg写入Channel内部维护的一个ChannelOutboundBuffer中，这个buffer内部持有一个链表结构的Entry，每次write都会将msg封装成Entry并添加到链表的tail。当调用flush()方法时才会触发实际的socket写入操作。此处需要注意，write操作将会依次执行pipeline中添加的所有的Encoder。
 *  void flush()：将当前Channel持有的ChannelOutboundBuffer中尚未flush的Entry写入Socket。通常，我们尽可能在write一定量的数据后立即调用flush。如果遗忘flush或者较少次数的调用，将会导致内存过度消耗。
 *  ChannelFuture writeAndFlush(Object msg)：同上，只是在write结束后，立即flush。通常建议使用此方法。write和writeAndFlush方法都会依次经过ChannelOutboundHandler的write方法（Encoder）。
 *  Channel read()：此方法并没有返回ChannelFuture，这个方法主要作用就是在Selector上注册此Channel的READ事件，内部并没有发生实际的read操作。开发工程师通常不会直接调用此方法，这个方法或许是框架使用的方法。在创建NioSocketChannel、NioServerSocketChannel时，可以指定一个autoRead参数，表明当Channel创建后，是否立即注册READ事件，默认为“true”，那么将立即调用此read方法；如果开发者设定了false，那么开发者则需要在合适的实际（比如channelActive时）手动调用此read方法。此方法将会依次执行pipeline中ChannelOutboundHandler的read方法。当NIO Channel中有实际数据读取时，才会触发ChannelInbountHandler的channelRead()方法，这是channelRead和read方法的区别。

所谓异步，就是操作的执行结果将不会在操作本体结束后立即获得，比如write操作，当write方法返回时并不表示数据已经通过底层Socket已经发送成功。Netty将异步操作封装成ChannelFuture，我们可以在future中查看操作的状态：success、cancell还是抛出了异常。我们稍后介绍Future。

Channel是有“继承”关系的，它有一个parent()方法用来获取其“父Channel”，所谓parent，就是创建此Channel的Channel，比如SocketChannel是由ServerSocketChannel创建，那么SocketChannel的parent就是ServerSockentChannel。

Channel接口实现类AbstractChannel，实现了基本的功能。比较常接触到的子类有NioSocketChannel、NioServerSocketChannel，这两个API我们通过上述的列子已经得知，对于ServerBootstrap将使用NioServerSocketChannel（底层通过反射机制创建其实例），对于Bootstrap则使用NioSocketChannel。

Channel接口中还包含了一个子接口：Unsafe；这个Unsafe和JDK自带的Unsafe没有关系，Netty中的Unsafe只是框架内部使用，主要用来操作底层的Socket，比如connect、close、read、write；即Channel接口中有关底层Socket操作的，将会有Unsafe来操作javaChannel()即可。到此为止，我们已经解开了Channel内部的设计原理。

**三、ChannelFuture接口：异步操作结果。**

这个接口和java中的Future所描述的语义是一致的，大家在开发线程池有关的应用时，应该对Future、ScheduledFuture有所了解，它表达了当Channel IO异步操作的结果。Netty中Channel IO都是异步的，即IO调用方法立即返回，但此时并不表示IO实际操作已经实际执行结束，只是返回一个ChannelFuture实例，具体执行的结果状态可以通过检测Future才能得到。ChannelFuture继承自java.util.concurrent.Future。

对于一个ChannelFuture实例，执行的结果有2种状态：completed、uncompleted。uncompleted表示IO正在执行尚未结束；一个completed的IO操作，将由三个结果：succeeded、failed、cancelled，这个很好理解。ChannelFuture继承了java.util.concurrent.Future，额外提供了一些方法：

*  boolean isDone()：操作是否完成，completed 还是 uncompleted
 *  boolean isCanclled()：如果Future已经完成，则判断操作是否被取消。
 *  boolean isSuccess()：同上。
 *  Throwable cause()：如果执行失败，此处可以获取导致失败的exception信息。
 *  ChannelFuture await()：等待，直到异步操作执行完毕，内部基于wait实现。
 *  ChannelFuture sync()：等待，直到异步操作执行完毕，核心思想同await。我们得到Future实例后，可以使用sync()方法来阻塞当前线程，直到异步操作执行完毕。和await的区别为，如果异步操作失败，那么将会重新抛出异常（将上述cause()方法中的异常抛出）。await和sync一样，当异步操作执行完毕后，通过notifyAll()唤醒。
 *  ChannelFuture addListener(GenericFutureListener listener)：向Future添加一个listener，当异步操作执行完毕后（无论成败），会依次调用listener的operationCompleted方法。

Netty Future中并没有提供类似于notify的接口以供开发者调用（不过Object有此方法，但开发者不可以调用），即唤醒阻塞全权交给内部实现（Future的执行完毕后）。

Netty声明，不要在IO线程中调用sync()、await()等相关阻塞的方法，这可能会带来死锁问题。我们上述所介绍的各种ChannelHandler都会在Netty IO线程中调用，所以不要在channelRead、write方法中调用Future的阻塞方法；Netty推荐使用addListener这种异步的方式来解决。

Java代码   ![收藏代码][icon_star.png]

1.  ChannelFuture future = ctx.channel().close();  
2.  future.addListener(**new** ChannelFutureListener() \{  
3.      @Override  
4.      **public** **void** operationComplete(ChannelFuture future) **throws** Exception \{  
5.          //  
6.      \}  
7.  \});

如果需要对IO操作进行同步，那么sync方法是比较合适的选择，通常这个方法需要在非IO线程中调用。

GenericFutureListener有个子接口：ChannelFutureListener，此接口有3个比较常用的已经实现的子类，比如ChannelFutureListener.CLOSE表示当future执行完毕后关闭channel。

**四、EventLoopGroup：**

上述的例子中我们看到2个API：Bootstrap和ServerBootStrap，Bootstrap用于初始化Netty Client端，ServerBootstrap用于Netty Server端，这对于Socket编程而言是通例。

EventLoopGroup涉及到Netty的线程模型，如果你了解NIO中常用的reactor模型，那么理解Netty模型将非常容易的，我们可以简单的认为EventLoopGroup是一个线程池调度服务，这和我们常用的FixedThreadPool在设计思想上没有太大区别；在Netty中比较常用的子类就是NioEventLoopGroup，它继承了ScheduledExecutorService。我们在创建NioEventLoopGroup时，可以指定线程池的容量，默认为：CPU处理器个数 \* 2。

需要提醒的是，NioEventLoopGroup线程池的实现和java中提供的线程池有所不同，内部数据结构使用数组来表示“线程池”（java中的线程池为hashSet，元素值为Thread类型，这种结构是为了适应多种类型线程池的设计），数组的每个元素类型为NioEventLoop，当NioEventLoopGroup创建时将会根据线程池的容量创建每个NioEventLoop实例。

Java代码   ![收藏代码][icon_star.png]

1.  **for** (**int** i = 0; i < nThreads; i ++) \{  
2.      **boolean** success = **false**;  
3.      **try** \{  
4.          children\[i\] = newChild(threadFactory, args);//create a NioEventLoop  
5.          success = **true**;  
6.      \} **catch** (Exception e) \{  
7.          ///  
8.      \}  
9.      ...  
10. \}

NioEventLoop继承了SingleThreadEventExecutor，这个Executor也继承了ExecutorService，简单而言，NioEventLoop也是一个“线程池调度服务”，只是它内部只有一个Thread，在创建NioEventLoop实例时，其内部会创建一个的Thread实例以及一个用来保存task的LinkedBlockingQueue，这个内部Thread是延迟启动的，即只有当此NioEventLoop队列中有task提交时才会启动。

NioEventLoop实例创建时，同时会创建一个Selector（通过SelectorProvider.open()），即每个NioEventLoop都持有一个selector实例，由此可见，NioEventLoopGroup的线程池容量，就是线程的个数，也是Bootstrap中持有的Selector的数量。每个NioEventLoop实例内部Thread负责select多个Channel的IO事件（NIO Selector.select），如果某个Channel有事件发生，则在内部线程中直接使用此Channel的Unsafe实例进行底层实际的IO操作。简单而言，就是让每个NioEventLoop管理一组Channel。

对于ServerBootstrap而言，创建2个NioEventLoopGroup，其中“bossGroup”为Acceptor 线程池，这个线程池只需要一个线程（大于1，事实上没有意义），它主要是负责accept客户端链接，并创建SocketChannel，此后从“workerGroup”线程池（reactor）中以轮询的方式(next)取出一个NioEventLoop实例，并将此Channel注册到NioEventLoop的selector上，此后将由此selector负责监测Channel上的读写事件（并由此NioEventLoop线程负责执行）。由此可见，对于ServerBootstrap而言，bossGroup中的一个线程的selector只关注SelectionKey.OPT\_ACCEPT事件，将接收到的客户端Channel绑定到workerGroup中的一个线程，全局而言ServerSocketChannel和SocketChannel并没有公用一个Selector，ServerSocketChannel单独使用一个selector（线程），而众多SocketChannel将会被依次绑定到workerGroup中的每个Selector；这在高并发环境中非常有效，每个Selector响应事件的及时性更强，如果Selector异常(比如典型的空轮询的BUG,即Select方法唤醒后缺得到一个空的key列表，而死循环下去，CPU空转至100%，这是由于Epoll的BUG引起)，只需要rebuild当前Selector即可（Nettyselect唤醒后，如果没有获取到事件keys列表且这种空唤醒的次数达到阀值），影响的Channel数量比较有限。同时这也是为什么开发者不能在IO线程中使用阻塞方法(比如wait)的原因，同一个Selector下的所有Channel公用一个线程，这种阻塞其实会导致当前线程下其他Channel（包括当前Channel）中后续事件的select无法进行，因为一个Selector中的所有selectionKey都会在此线程中依次执行。

事实上，我们在创建ServerBootstrap时可以不指定workerGroup，即为null；那么Acceptor和reactor都将公用一个线程池，这在原则上并不错，通常也不会带来问题，但是当Client创建链接比较频繁的应用中、业务处理耗时时，会加剧Acceptor线程的延迟性（因为这个线程有可能还服务与其他SocketChannel），通常我们建议为ServerBootstrap创建两个NioEventLoopGroup。

对于Bootstrap而言，问题就简单很多，一个Bootstrap对应一个Client端SocketChannel，即workerGroup中只需要一个线程即可（而且只有一个线程处于服务状态）。通常我们可以通过创建多个Bootstrap（Bootstrap Pool）实例即创建多个Socket连接，来提升Client端整体的吞吐能力。

NioEventLoopGroup有一个方法是和NIO有关的：register(Channel channel)，此方法的主要作用就是将Channel注册到NioEventLoop上selector中，这个方法开发者通常不会调用。当ServerBootstrap.bind操作执行成功后或者BootStrap.connect操作执行成功后，都会触发register，即将当前java Channel注册到Selector上，这一点和NIO的设计保持一致。register方法主要是获取一个SelectionKey，这个key将会由当前Channel持有，当Channel处于可用状态后（active），才会通过此key注册感兴趣的事件（比如ServerSocketChannel注册ACCEPT事件，SocketChannel注册READ事件）。

**五、Pipeline与ChannelHandler**

在上述例子中，ServerBootstrap需要通过childHandler()方法指定，当SocketChannel创建时将会把此值传递给SocketChannel；对于Bootstrap通过handler()方法指定，因为Client不会创建所谓的“child”。我们例子中使用了ChannelInitializer，它是一个特殊的handler，会在Channel实例化时被首先添加到pipeline中，当Channel注册成功后，NioEventLoop线程会依次执行pipeline中的所有handler中的channelRegistered方法（也就是这个ChannelInitializer），这个方法会调用ChannelInitializer的initChannel，即Netty给我们提供了这种便捷的方式向Channel添加自定义的Handler列表。事实上开发者可以在任何需要的时候向pipeline中添加handler，不过通过ChannelInitializer方式可以一蹴而就。

pipeline在Channel初始化时创建，内部是基于链表实现（ChannelPipeline），即有head和tail，其addLast方法将会把自定义handler添加到tail之前，addFirst方法将会handler添加到Head之后。也可以在运行时remove掉某个handler，pipeline中的handler不能重复，默认由class名称来限定。head和tail是Netty内置的2个ChannelHandler，开发者不能移除，Netty将通过它们进行一些内部操作。每个Channel都有自己的pipeline，但是所有Channel公用handler实例（通过handler()、childHandler()指定的那个实例被公用），那也意味着在handler实例中使用公用的变量是不明智的。不过像ChannelInitializer这样的Handler，在initChannel方法中创建（new）新的handler并添加到Channel的pipleline中，这些新的handler只会被当前Channel持有，那么在这些handler中使用公用变量不会有问题，也不会出现并发问题，在解决“半包”问题时我们就通常在某个handler中使用公用变量来保存数据直到“封包”。

ChannelHandler有2种：ChannelInboundHandler和ChannelOutboundHandler，从字面意思上说，inbound即为入站，outbound为出站；对于Netty通道而言，从Socket通道中read数据进入Netty Handler的方向为inbound，从Netty handler向Socket通道write数据的方向为outbound。上述例子，我们也看到，Encoder为outbound，Decoder为inbound。简而言之，Socket中read到数据后，将会从tail到head依次执行链表中的inbound handler实例的相关方法（比如channelRead，channelReadComplete方法）；当开发者通过channelHandlerContext向Socket写入数据时，将会从head到tail方向依次执行链表中的outbound handler实例的相关方法（比如write）。pipeline中的Head Handler是一个outbound实例，它处于outbound方向的最后一站，由此可见它必须处理那些“bind”、“connect”、“read”（非read数据，而是向Channel注册感兴趣的事件，在channel注册成功后执行）操作，以及使用unsafe操作Socket写入实际数据；Tail Handler是一个inbound实例，它是Socket read到数据后，第一个交付的inbound，不过从源码来看，tail似乎并没有做什么实质性的操作。

ChannelHandler的实现类比较多，对于开发者而言，只需要知道Netty已经提供了大多数Encoder、decoder即可，它们为我们提供了一些基本的解决方案，在此处不再赘言。对于开发者而言，只需要实现ChannelInboundHandlerAdapter、ChannelOutboundHandlerAdapter这种子类即可。Inbound、outbound接口中的所有方法都会被ChannelPipeline在合适的时机被触发，pipeline中有各种对应的fire方法。

ChannelInboundHandler接口中几个常用方法介绍：

*  void channelRegistered(ChannelHandlerContext ctx): 当channel注册成功后执行，即channel绑定到NioEventLoop上，且将Channel注册到selector之后执行。不过此时channel尚不能进行实际的操作。
 *  void channelActive(ChannelHandlerContext ctx): channel注册首次成功后执行，即channelRegistered方法执行后调用；如果Netty开启了autoRead配置项（默认为true），那么内置的handler将会在channelActive方法中向Channel注册默认的NIO 事件，比如ServerSocketChannel注册ACCEPT，SocketChannel注册READ；对于开发者而言，此时channel已经可用，我们可以在此方法中执行一些与通道有关的初始化操作，比如向Socket写入一些“handshake”信息等。
 *  void channelRead(ChannelHandlerContext ctx,Object msg): socket将已经读取到数据传递给handler，此方法在NIO中READ事件触发后并读取到任意字节数据后被fire；这个方法是我们经常接触的。
 *  void channelReadComplete(ChannelHandlerContext ctx): read操作结束后调用；read数据首先被添加到Bytebuf中，如果这个Bytebuf已满，则会终止read操作，调用此方法。（底层使用javaChannel().read(ByteBuffer)）。
 *  void channelInactive(ChannelHandlerContext ctx): 当channel被Close后，调用此方法。
 *  void channelUnregistered(ChannelHandlerContext ctx): 当channel的selectionKey被取消后执行，通常时在Channel被关闭时。

ChannelOutboundHandler接口中几个常用的方法：

*  void read(ChannelHandlerContext ctx): 这个read方法和inbound中的channelRead()方法不同，此方法通常会在channelActive之后调用，表示channel即将注册感兴趣的NIO事件，并不会真正read数据；开发者通常可以在此方法中做一些准备工作。
 *  void write(ChannelHandlerContext ctx,Object msg,ChannelPromise promise): promise为ChannelFuture的子接口，表示异步操作的结果；通过此方法向pipeline中写入数据，msg表示正在写入的数据，其实数据并没有真正写入到Socket中，而是被缓存起来，直到调用flush。
 *  void flush(ChannelHandlerContext ctx):当调用ChannelHandlerContext.flush()方法时执行，表示数据即将被flush到Socket中。

至此，我们已经清楚了如下流程：Channel创建后，初始化其pipeline，当Channel注册到Selector后（包括绑定到NioEventLoop线程），执行ChannelInitializer这个特殊的handler并将开发者指定的其他多个handler添加到pipeline中；当NioEventLoop线程中，selector检测到NIO事件后，将会依次执行此Channel pipeline中的相应的fire方法（比如fireChannelRead()）,那么fire方法将会从head或者tail这两个特殊的handler开始，依次调用pipeline中其他handler的实际方法（比如channelRead()）。

需要注意：pipeline中的某个handler认为操作需要继续继续传递下去，那么在此handler的实际业务执行完毕之后要通过调用ChannelHandlerContext相应的fire方法，将事件继续传递给下一个handler；如果没有调用fire方法，那么此事件将会被终止。比如某个handler（不是最后一个）的channelRead()方法中没有调用ChannelHandlerContext.fireChannelRead()，那么后续的其他handler将不会被执行。所以开发者可以在自己的handler中决定事件是否继续传播下去，在Netty提供的各种Decoder中我们可以看到这种例子，比如“FixedLengthFrameDecoder”直到buffer中可读数据的长度达到配额后才会将事件传播给下一个handler。

在Netty-handler包中还额外的提供了其他的辅助性handler，比如IdleStateHandler（链接空闲超时后怎么处理）、ChunkedWriteHandler（封包写入）、LoggingHandler（控制日志写入）、ChannelTrafficShapingHandler（流量整形、流量控制）等等，在某些特殊业务场景下非常有用。

**六、ByteBuf**

和NIO中的ByteBuffer类语义一样，用于保存一序列字节数组（byte array），可以对buffer中的数据进行read、write操作；在整体设计上与NIO ByteBuffer也极为相似，比如buffer内部基于字节数组实现，在ByteBuffer中有三个配额数字：capacity（总容量）、limit（max readIndex）、position（write/read index）；不过在ByteBuf中为：readerIndex（read操作起始index，递增）、writeIndex（类似于position）、capacity。ByteBuffer没有严格区分出readIndex和writeIndex，所以ByteBuffer通常不能混合使用read和write操作，因为write和read都会使用position字段标记操作的index，所以通常为“写完再读”或者“读完重写”（回忆一下flip方法）；对于ByteBuf而言，事情有些不同，writeIndex表示写入数据的起始位置，readIndex表示读取数据的起始位置，readIndex <= writeIndex < capacity，writeIndex起到了read limit的作用，那么一个ByteBuf写入数据时只会导致writeIndex增加，不影响后续的read操作（没有读完的话，可以继续read，也可以write），所以ByteBuf比NIO ByteBuffer设计的更加优秀。ByteBuf与ByteBuffer之间可以互相转换，它们共享底层的字节数组，比如ByteBuf.nioBuffer()获取一个ByteBuffer实例，此实例与ByteBuf共享可读字节数组(readIndex -> writeIndex之间)。

此外ByteBuf提供了discardReadBytes()方法，即将那些已经读取的数据清除，将那些尚未读取的数据copy移动到字节数据的前端，最终可读取的数据量没有变化，可写入的数据容量增加，readIndex置为0。这个方法可以帮助开发者在不需要重新创建ByteBuf实例的情况下，重用buffer，对性能有一定的提升（底层仍会数组copy）。clear方法和ByteBuffer中的clear语义一致，将数据的数据清空（逻辑），readIndex和writeIndex均置为0。

Netty中，ByteBuf根据其内存分配机制，有2种实现：HeapByteBuf和DirectByteBuf（直接内存分配），这一点和NIO一样。Netty为了提升buffer的使用效率，减少因分配内存和内存管理带来的性能开支，又将ByteBuf分为：Pooled和Unpooled两类，即是否将ByteBuf基于对象池。最终ByteBuf子类为：PooledHeapByteBuf，PooledDirectByteBuf，UnpooledDirectByteBuf，UnpooledHeapByteBuf。不过和NIO ByteBuffer不同的时，ByteBuf并没有提供视图类，比如CharBuffer、IntBuffer等。

HeapByteBuf：同NIO种的HeapByteBuffer，即堆内存buffer，内存的创建和回收速度较快，缺点是内存Copy，即当HeapByteBuf的数据需要与Socket（或者磁盘）进行数据交换时，需要将数据copy到相应的驱动器缓冲区中，效率稍低。

DirectByteBuf：直接内存，或者说堆外内存，有操作系统分配，所以分配和销毁的速度稍慢，一般适用于“使用周期较长、可循环使用”的场景，优点是进行跨介质数据交换时无需数据Copy，速度稍高一些。

通常Socket数据直接操作端使用DirectByteBuf，上层业务处理阶段（比如编解码）可以使用HeapByteBuf。

ByteBuffer还有一个比较不友好的设计：容量不可变；即一个ByteBuffer在创建时需要指定容量（ByteBuffer.allocate(capacity)），而且将会立即申请此capacity大小的字节数组空间，此后capacity不可调整。ByteBuf则稍有不同，它在创建时指定一个maxCapacity（最大容量），不过并不会立即申请此容量大小的数组空间，当首次write数据时，才会初始一定容量的空间，此后空间动态调整直到达到maxCapacity；如果ByteBuf中已分配（或者write需要）的容量小于4M时，首先分配64个字节，此后以double的方式扩容(128,512,1024...)直到4M为止，当容量达到4M后，当空间不足时每次递增4M（而不是double）；以4M为分界（为什么是4M？），采取2中扩容手段，不仅能够提高扩容效率（<4M）,而且可以避免内存盲目消耗（> 4M）。

我们会发现ByteBuf继承了一个ReferenceCounted接口，上述我们谈到的ByteBuff的子类都实现了AbstractReferenceCountedByteBuf，为什么Netty要与“引用计数器”产生关系？JVM中使用“计数器”（一种GC算法）标记对象是否“不可达”进而收回，Netty也使用了这种手段来对ByteBuf的引用进行计数，如果一个ByteBuf不被任何对象引用，那么它就需要被“回收”，这种“回收”可能是放回对象池（比如Pooled ByteBuf）或者被销毁，对于Heap类型的ByteBuf则直接将底层数组置为null，对于direct类型则直接调用本地方法释放外部内存（unsafe.freeMemory）。Netty采用“计数器”来追踪ByteBuf的生命周期，一是对Pooled ByteBuf的支撑，二是能够尽快的“发现”那些可以回收的ByteBuf（非Pooled），以便提成ByteBuf的分配和销毁的效率。（参见：[“引用计数器”][Link 1]）

ReferenceCounted接口中，有几个方法：

*  int refCnt()：获取引用计数。一个新创建的对象，起始为1。
 *  ReferenceCounted retain()：增加一次引用计数。
 *  boolean release()：释放一次引用计数，如果此时计数器值为0，表示没有它没有被其他任何对象引用，则“deallocate”（回收）此对象。可以使用ReferenceCountUtil来释放引用计数。

我们通常这样：

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** send(ByteBuf byteBuf) \{  
2.      **try** \{  
3.          //do something;but not sent the byteBuf to next handler;  
4.      \} **finally** \{  
5.          byteBuf.release();  
6.      \}  
7.  \}

1）要么让ByteBuf最后一个处理者去release，ByteBuf可以在多个方法之间传递，直到最终某个方法决定此ByteBuf不需要继续被处理，然后release，上层调用者一旦将ByteBuf传递出去之后将不能再次使用此ByteBuf。就像inbound handler一样，中间所有的handler处理完ByteBuf之后直接传递给下一个，由最后一个handler负责release。这也要求开发者自定义的handler遵循这个原则。

2）要么，retain和release方法同时出现，即将ByteBuf传递给下一个方法之前，首先retain，传递之后立即release，且接收ByteBuf的方法也这么做。

Java代码   ![收藏代码][icon_star.png]

1.  **public** **void** main() \{  
2.      ByteBuf byteBuf = Unpooled.buffer(128);  
3.      **try** \{  
4.          methodA(byteBuf);  
5.      \} **finally** \{  
6.          byteBuf.release();  
7.      \}  
8.    
9.  \}  
10.   
11. **public** **void** methodA(ByteBuf byteBuf) \{  
12.     byteBuf.retain();  
13.     **try** \{  
14.         methodB(byteBuf);  
15.     \} **finally** \{  
16.         byteBuf.release();  
17.     \}  
18. \}

在Netty中，Unpooled ByteBuf是一种比较简单的buf，它和NIO中的ByteBuffer设计和实现基本一样：“使用完即可销毁”，通常无需多次复用，实现简单，维护成本较低，不易出错，在满足性能要求的情况下，我们通常优先选择Unpooled ByteBuf。在Unpooled类中提供了多个static方法，来帮助我们创建需要的ByteBuf实例。

Java代码   ![收藏代码][icon_star.png]

1.  ByteBuf byteBuf = Unpooled.buffer(1024);  
2.  //或者  
3.  ByteBuf byteBuf = UnpooledByteBufAllocator.DEFAULT.newHeapBuffer(1024);  
4.  //对于开发者，通常为  
5.  ByteBuf byteBuf = channelHandlerContext.alloc().buffer();

PooledByteBuf即“对象池”化的ByteBuf，在此之前，我们需要先了解一下[“Memory arena”][Memory arena]：为了集中管理应用程序对内存的分配，将首先申请一块较大的连续的内存空间--Arena，并将此空间有层级的切分为多个chunk，每个chunk又被分为更小粒度的pages（不同的实现中，可能各个chunk的容量不一定相同），此后应用程序申请内存空间时，只需要从arena中找到合适的未被使用的chunk或者pages即可（如果一个Arena不足，则可以创建多个，弹性设计），如果空间释放，只需要将它们标记为“未使用”，因为内存的申请和释放交替进行，最终整个生命周期中，所有的内存操作只需要在一个或者几个Arena之间（包括chunks）“调配”即可，它直接提升了内存分配和回收的效率，对性能的提升非常直接。（不过Netty 4仍然将Unpooled作为默认的内存分配器）

那么PooledByteBuf，设计思想类似于“Memory Arena”（内部实现为PoolArena）： 一个PoolArena有多个chunk组成的大内存区域，每个chunk又多个pages组成，每个page又被分为大小相等的多个subPages（不过不同的page下的subPage大小可能不同，subPage大小由初次申请page时决定，比如此page首次被申请1K的buf，那么此page将有8个1K的subpages组成），pageSize默认为8192（即8K），每个chunk最大为16M（8192 << 11）,每个Arena默认有3个chunk，全局所有的Arena占用的总内存不得超过JVM内存总量的 50%，根据内存类型又分为HeapArena和DirectArena，在PooledByteBufAllocator初始化时即创建一定个数的HeapArena和DirectArena（数组结构，HeapArena\[\]，并不实际占用空间，数组大小由内存50%计算出来的）；内部原理比较复杂，分配内存时会根据required大小（<512，<pageSize，<chunkSize各有不同的分配策略），从合适的subpages中找到足够多的尚未分配的空间，然后封装成ByteBuf返回。回收时，只需要将PooledByteBuf对应的chunk上的subpage标记为free即可。源码参见：PoolArena，PooledByteBuf，PooledByteBufAllocator。

Java代码   ![收藏代码][icon_star.png]

1.  ByteBuf byteBuf = PooledByteBufAllocator.DEFAULT.heapBuffer();

对于开发者而言，channelHandlerContext.alloc()返回的是“默认配置”的ByteBufAllocator，即使用“Unpooled”内存模型和“heap”内存。我们可以通过“-Dio.netty.allocator.type=pooled”指定默认使用“Pooled”模型，“-Dio.netty.noPreferDirect=false”（默认为false）和“-Dio.netty.noUnsafe=true”（默认为false）来指定默认使用direct内存。

事实上，如果开发者没有绝对的基准测试，建议不要随意改动netty的内部调优参数，有可能会带来不可预知的性能问题。

**七、总结**

1、Netty是一个NIO异步框架，高度封装了java NIO，并提供了一套规范性的编码，开发者可以快速的开发出比较健壮而高性能的应用。

2、Netty通过线程池模型和基于Pipeline机制的handler，可以让Netty框架更加具有扩展性。

3、Netty内置了多种handler，来解决“封包”、“成帧”等问题。

4、Netty通过重实现ByteBuf，以及高性能的Pooled特性，对开发者而言，处理buffer相关的数据更加有效。

5、Netty已经成为Protobuf、Avro、Thrift等多中RPC框架的底层通道。

转载于：http://shift-alt-ctrl.iteye.com/blog/2219057

[icon_star.png]: /images/20220604/e112379078194effa2dd5cd0ee45c437.png
[Link 1]: http://netty.io/wiki/reference-counted-objects.html
[Memory arena]: https://en.wikipedia.org/wiki/Region-based_memory_management