Netty核心知识总结（含部分源码解析）

前言

基本介绍

Netty是一个高性能、异步事件驱动的NIO框架，它提供了对TCP、UDP和文件传输的支持，作为一个异步NIO框架，Netty的所有IO操作都是异步非阻塞的，通过Future-Listener机制，用户可以方便的主动获取或者通过通知机制获得IO操作结果。

作为当前最流行的NIO框架，Netty在互联网领域、大数据分布式计算领域、游戏行业、通信行业等获得了广泛的应用，一些业界著名的开源组件也基于Netty的NIO框架构建。

GitHub：https://github.com/netty/netty

官网：https://netty.io/

「Netty的优缺点」

之前我们使用JAVA NIO的时候会有一些问题：

• JDK NIO的BUG，例如臭名昭著的epoll bug，它会导致Selector空轮询，最终导致CPU 100%，官方声称在JDK 1.6版本的update18修复了该问题，但是直到JDK 1.7版本该问题仍旧存在，只不过该BUG发生概率降低了一些而已，它并没有得到根本性解决。
• 客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常码流的处理等问题需要处理。

Netty的优点：

1. API使用简单，开发门槛低;
2. 功能强大，预置了多种编解码功能，支持多种主流协议;
3. 定制能力强，可以通过ChannelHandler对通信框架进行灵活地扩展;
4. 性能高，通过与其他业界主流的NIO框架对比，Netty的综合性能最优;
5. 成熟、稳定，Netty修复了已经发现的所有JDK NIO BUG，业务开发人员不需要再为NIO的BUG而烦恼;
6. 社区活跃，版本迭代周期短，发现的BUG可以被及时修复，同时，更多的新功能会加入;
7. 经历了大规模的商业应用考验，质量得到验证。

Netty有一个最重要的缺点：大版本不兼容，3.x/4.x同时维护，5.x放弃维护，主要原因是Netty抛弃了Jboss单独发展了。

「Netty的应用产品」

Netty 经过很多出名产品在线上的大规模验证，其健壮性和稳定性都被业界认可，其中典型的产品有一下几个

• 服务治理：Apache Dubbo、GRPC。
• 大数据：Hbase、Spark、Flink、Storm。
• 搜索引擎：Elasticsearch。
• 消息队列：RocketMQ、ActiveMQ。

Spring WebFlux是伴随Spring framework 5提出的网络框架解决方案，也是基于Netty实现。

还有更多优秀的产品可以参考下面网址：https://netty.io/wiki/related-projects.html。

入门案例

构建一个简单的应用程序：客户端将消息发送给服务器，而服务器再将消息会送给客户端。

首先，引入 Maven 依赖

1. <dependency> 
2.        <groupId>io.netty</groupId> 
3.        <artifactId>netty-all</artifactId> 
4. /dependency> 

「服务端：」

Netty 服务器都需要以下两部分：

• 至少一个ChannelHandler，该组件实现了服务器对从客户端接收的数据的处理，即它的业务逻辑。
• 配置引导服务器的启动代码，它会将服务器绑定到它要监听连接请求的端口上。

「ChannelHandler服务端业务逻辑」

因为你的 Echo 服务器会响应传入的消息，所以它需要实现 ChannelInboundHandler 接口，用来定义响应入站事件的方法。

这个简单的应用程序只需要用到少量的这些方法，所以继承 ChannelInboundHandlerAdapter 类也就足够了，它提供了 ChannelInboundHandler 的默认实现。

我们需要的方法是:

• channelRead()，对于每个传入的消息都要调用;
• channelReadComplete()，通知ChannelInboundHandler最后一次对channelRead()的调用是当前批量读取中的最后一条消息;
• exceptionCaught()，在读取操作期间，有异常抛出时会调用。

该 Echo 服务器的 ChannelHandler 实现是 EchoServerHandler：

「引导服务器」

引导过程中所需要的步骤如下:

• 创建一个 ServerBootstrap 的实例以引导和绑定服务器;
• 创建并分配一个 NioEventLoopGroup 实例以进行事件的处理，如接受新连接以及读/写数据;
• 指定服务器绑定的本地的 InetSocketAddress;
• 使用一个 EchoServerHandler 的实例初始化每一个新的 Channel;
• 调用 ServerBootstrap.bind()方法以绑定服务器。

「客户端：」

Echo 客户端将会:

• 连接到服务器;
• 发送一个或者多个消息;
• 对于每个消息，等待并接收从服务器发回的相同的消息;
• 关闭连接。

编写客户端所涉及的两个主要代码部分也是业务逻辑和引导

「ChannelHandler实现客户端逻辑」

如同服务器，客户端将拥有一个用来处理数据的 ChannelInboundHandler。

在这 个场景下，你将扩展 SimpleChannelInboundHandler 类以处理所有必须的任务。

这要求重写下面的方法:

• channelActive()——在到服务器的连接已经建立之后将被调用;
• channelRead0()——当从服务器接收到一条消息时被调用;
• exceptionCaught()——在处理过程中引发异常时被调用。

「引导客户端」

• 为初始化客户端，创建了一个 Bootstrap 实例;
• 为进行事件处理分配了一个 NioEventLoopGroup 实例，其中事件处理包括创建新的连接以及处理入站和出站数据;
• 为服务器连接创建了一个 InetSocketAddress 实例;
• 当连接被建立时，一个 EchoClientHandler 实例会被安装到ChannelPipeline 中;
• 在一切都设置完成后，调用 Bootstrap.connect()方法连接到远程节点;

基本架构

Netty 的逻辑处理架构为典型网络分层架构设计，共分为网络通信层、事件调度层、服务编排层，每一层各司其职。

网络通信层

网络通信层的职责是执行网络 I/O 的操作，它支持多种网络协议和 I/O 模型的连接操作。

当网络数据读取到内核缓冲区后，会触发各种网络事件，这些网络事件会分发给事件调度层进行处理。

网络通信层的核心组件包含BootStrap、ServerBootStrap、Channel三个组件。

「BootStrap和ServerBootStrap」

Bootstrap 是引导的意思，它主要负责整个 Netty 程序的启动、初始化、服务器连接等过程，它相当于一条主线，串联了 Netty 的其他核心组件。

Netty 中的引导器共分为两种类型：一个为用于客户端引导的 Bootstrap，另一个为用于服务端引导的 ServerBootStrap，它们都继承自抽象类 AbstractBootstrap。

Bootstrap 和 ServerBootStrap 十分相似，两者非常重要的区别在于 Bootstrap 可用于连接远端服务器，只绑定一个EventLoopGroup。

ServerBootStrap 则用于服务端启动绑定本地端口，会绑定两个 EventLoopGroup，这两个 EventLoopGroup 通常称为 Boss 和 Worker。

「Channel」

Channel 的字面意思是通道，它是网络通信的载体。

Channel提供了基本的 API 用于网络 I/O 操作，如 register、bind、connect、read、write、flush 等。

Netty 自己实现的 Channel 是以 JDK NIO Channel 为基础的，相比较于 JDK NIO，Netty 的 Channel 提供了更高层次的抽象，同时屏蔽了底层 Socket 的复杂性，赋予了 Channel 更加强大的功能，在使用 Netty 时基本不需要再与 Java Socket 类直接打交道。

Channel 会有多种状态，如连接建立、连接注册、数据读写、连接销毁等。

随着状态的变化，Channel 处于不同的生命周期，每一种状态都会绑定相应的事件回调，下面的表格列举了 Channel 最常见的状态所对应的事件回调。

事件调度层

事件调度层的职责是通过 Reactor 线程模型对各类事件进行聚合处理，通过 Selector 主循环线程集成多种事件( I/O 事件、信号事件、定时事件等)，实际的业务处理逻辑是交由服务编排层中相关的 Handler 完成。

事件调度层的核心组件包括 EventLoopGroup、EventLoop。

「EventLoopGroup和EventLoop」

EventLoopGroup 本质是一个线程池，主要负责接收 I/O 请求，并分配线程执行处理请求。

EventLoopGroup、EventLoop、Channel 的关系：

• 一个 EventLoopGroup 往往包含一个或者多个 EventLoop。
• EventLoop 用于处理 Channel 生命周期内的所有 I/O 事件，如 accept、connect、read、write 等 I/O 事件。
• EventLoop 同一时间会与一个线程绑定，每个 EventLoop 负责处理多个 Channel。
• 每新建一个 Channel，EventLoopGroup 会选择一个 EventLoop 与其绑定，该 Channel 在生命周期内都可以对 EventLoop 进行多次绑定和解绑。

Netty 通过创建不同的 EventLoopGroup 参数配置，可以支持 Reactor 的三种线程模型：

• 单线程模型：EventLoopGroup 只包含一个 EventLoop，Boss 和 Worker 使用同一个EventLoopGroup;
• 多线程模型：EventLoopGroup 包含多个 EventLoop，Boss 和 Worker 使用同一个EventLoopGroup;
• 主从多线程模型：EventLoopGroup 包含多个 EventLoop，Boss 是主 Reactor，Worker 是从 Reactor，它们分别使用不同的 EventLoopGroup，主 Reactor 负责新的网络连接 Channel 创建，然后把 Channel 注册到从 Reactor。

服务编排层

服务编排层的职责是负责组装各类服务，它是 Netty 的核心处理链，用以实现网络事件的动态编排和有序传播。

服务编排层的核心组件包括 ChannelPipeline、ChannelHandler、ChannelHandlerContext。

「ChannelPipeline」

ChannelPipeline 是 Netty 的核心编排组件，负责组装各种 ChannelHandler，实际数据的编解码以及加工处理操作都是由 ChannelHandler 完成的。

ChannelPipeline 可以理解为ChannelHandler 的实例列表——内部通过双向链表将不同的 ChannelHandler 链接在一起。

当 I/O 读写事件触发时，ChannelPipeline 会依次调用 ChannelHandler 列表对 Channel 的数据进行拦截和处理。

ChannelPipeline 是线程安全的，因为每一个新的 Channel 都会对应绑定一个新的 ChannelPipeline。

一个 ChannelPipeline 关联一个 EventLoop，一个 EventLoop 仅会绑定一个线程。

「ChannelHandler和ChannelHandlerContext」

数据的编解码工作以及其他转换工作实际都是通过 ChannelHandler 处理的。

每创建一个 Channel 都会绑定一个新的 ChannelPipeline，ChannelPipeline 中每加入一个 ChannelHandler 都会绑定一个 ChannelHandlerContext。

ChannelHandlerContext 用于保存 ChannelHandler 上下文，通过 ChannelHandlerContext 我们可以知道 ChannelPipeline 和 ChannelHandler 的关联关系。

ChannelHandlerContext 可以实现 ChannelHandler 之间的交互，ChannelHandlerContext 包含了 ChannelHandler 生命周期的所有事件，如 connect、bind、read、flush、write、close 等。

核心流程

Netty 各个组件的整体交互流程：

• 服务端启动初始化时有 Boss EventLoopGroup 和 Worker EventLoopGroup 两个组件，其中 Boss 负责监听网络连接事件。当有新的网络连接事件到达时，则将 Channel 注册到 Worker EventLoopGroup。
• Worker EventLoopGroup 会被分配一个 EventLoop 负责处理该 Channel 的读写事件。每个 EventLoop 都是单线程的，通过 Selector 进行事件循环。
• 当客户端发起 I/O 读写事件时，服务端 EventLoop 会进行数据的读取，然后通过 Pipeline 触发各种监听器进行数据的加工处理。
• 客户端数据会被传递到 ChannelPipeline 的第一个 ChannelInboundHandler 中，数据处理完成后，将加工完成的数据传递给下一个 ChannelInboundHandler。
• 当数据写回客户端时，会将处理结果在 ChannelPipeline 的 ChannelOutboundHandler 中传播，最后到达客户端。

服务端启动流程

从服务端启动的一个非常精简的 Demo分析:

1. public class NettyServer { 
2.     public static void main(String[] args) { 
3.         NioEventLoopGroup bossGroup = new NioEventLoopGroup(); 
4.         NioEventLoopGroup workerGroup = new NioEventLoopGroup(); 
5.  
6.         ServerBootstrap serverBootstrap = new ServerBootstrap(); 
7.         serverBootstrap 
8.                 .group(bossGroup, workerGroup) 
9.                 .channel(NioServerSocketChannel.class) 
10.                 .childHandler(new ChannelInitializer<NioSocketChannel>() { 
11.                     protected void initChannel(NioSocketChannel ch) { 
12.                     } 
13.                 }); 
14.  
15.         serverBootstrap.bind(8000); 
16.     } 
17. } 

首先看到创建了两个NioEventLoopGroup，bossGroup表示监听端口，accept 新连接的线程组，workerGroup表示处理每一条连接的数据读写的线程组。

接下来我们创建了一个引导类 ServerBootstrap，这个类将引导我们进行服务端的启动工作。

我们通过.group(bossGroup, workerGroup)给引导类配置两大线程组。

然后，我们指定我们服务端的 IO 模型为NIO，我们通过.channel(NioServerSocketChannel.class)来指定 IO 模型，如果你想指定 IO 模型为 BIO，那么这里配置上OioServerSocketChannel.class类型即可。

接着，我们调用childHandler()方法，给这个引导类创建一个ChannelInitializer，这里主要就是定义后续每条连接的数据读写，业务处理逻辑。

之后在调用bind(8000)，我们就可以在本地绑定一个 8000 端口启动起来。

下面详细介绍：

「配置线程池」

Netty 是采用 Reactor 模型进行开发的，可以非常容易切换三种 Reactor 模式：单线程模式、多线程模式、主从多线程模式。

• 单线程模式

Reactor 单线程模型所有 I/O 操作都由一个线程完成，所以只需要启动一个 EventLoopGroup 即可。

1. EventLoopGroup group = new NioEventLoopGroup(1); 
2. ServerBootstrap b = new ServerBootstrap(); 
3. b.group(group) 

• 多线程模式

在 Netty 中使用 Reactor 多线程模型与单线程模型非常相似，区别是 NioEventLoopGroup 可以不需要任何参数，它默认会启动 2 倍 CPU 核数的线程，你也可以自己手动设置固定的线程数。

1. EventLoopGroup group = new NioEventLoopGroup(); 
2. ServerBootstrap b = new ServerBootstrap(); 
3. b.group(group) 

• 主从多线程模式

在大多数场景下，我们采用的都是主从多线程 Reactor 模型。

Boss 是主 Reactor，Worker 是从 Reactor。

它们分别使用不同的 NioEventLoopGroup，主 Reactor 负责处理 Accept，然后把 Channel 注册到从 Reactor 上，从 Reactor 主要负责 Channel 生命周期内的所有 I/O 事件。

1. EventLoopGroup bossGroup = new NioEventLoopGroup(); 
2. EventLoopGroup workerGroup = new NioEventLoopGroup(); 
3. ServerBootstrap b = new ServerBootstrap(); 
4. b.group(bossGroup, workerGroup) 

「Channel初始化」

• 设置 Channel 类型

推荐 Netty 服务端采用 NioServerSocketChannel 作为 Channel 的类型，客户端采用 NioSocketChannel。

1. b.channel(NioServerSocketChannel.class); 

Netty 提供了多种类型的 Channel 实现类，你可以按需切换，例如 OioServerSocketChannel、EpollServerSocketChannel 等。

「注册ChannelHandler」

1. b.childHandler(new ChannelInitializer<SocketChannel>() { 
2.     @Override 
3.     public void initChannel(SocketChannel ch) { 
4.         ch.pipeline() 
5.                 .addLast("codec", new HttpServerCodec()) 
6.                 .addLast("compressor", new HttpContentCompressor()) 
7.                 .addLast("aggregator", new HttpObjectAggregator(65536))  
8.                 .addLast("handler", new HttpServerHandler()); 
9.     } 
10. }) 

ServerBootstrap 的 childHandler() 方法需要注册一个 ChannelHandler。

ChannelInitializer是实现了 ChannelHandler接口的匿名类，通过实例化 ChannelInitializer 作为 ServerBootstrap 的参数。

Channel 初始化时都会绑定一个 Pipeline，它主要用于服务编排。

Pipeline 管理了多个 ChannelHandler。

I/O 事件依次在 ChannelHandler 中传播，ChannelHandler 负责业务逻辑处理。

上述 HTTP 服务器示例中使用链式的方式加载了多个 ChannelHandler，包含HTTP 编解码处理器、HTTPContent 压缩处理器、HTTP 消息聚合处理器、自定义业务逻辑处理器。

「设置Channel参数」

1. b.option(ChannelOption.SO_KEEPALIVE, true); 

ServerBootstrap 设置 Channel 属性有option和childOption两个方法，option 主要负责设置 Boss 线程组，而 childOption 对应的是 Worker 线程组。

「端口绑定」

在完成上述 Netty 的配置之后，bind() 方法会真正触发启动，sync() 方法则会阻塞，直至整个启动过程完成，具体使用方式如下：

1. ChannelFuture f = b.bind().sync(); 

客户端启动流程

客户端启动 Demo：

1. public class NettyClient { 
2.     public static void main(String[] args) { 
3.         NioEventLoopGroup workerGroup = new NioEventLoopGroup(); 
4.          
5.         Bootstrap bootstrap = new Bootstrap(); 
6.         bootstrap 
7.                 // 1.指定线程模型 
8.                 .group(workerGroup) 
9.                 // 2.指定 IO 类型为 NIO 
10.                 .channel(NioSocketChannel.class) 
11.                 // 3.IO 处理逻辑 
12.                 .handler(new ChannelInitializer<SocketChannel>() { 
13.                     @Override 
14.                     public void initChannel(SocketChannel ch) { 
15.                     } 
16.                 }); 
17.         // 4.建立连接 
18.         bootstrap.connect("aa.com", 80).addListener(future -> { 
19.             if (future.isSuccess()) { 
20.                 System.out.println("连接成功!"); 
21.             } else { 
22.                 System.err.println("连接失败!"); 
23.             } 
24.  
25.         }); 
26.     } 
27. } 

我们描述一下客户端启动的流程：

1. 首先，与服务端的启动一样，我们需要给它指定线程模型，驱动着连接的数据读写。
2. 然后，我们指定 IO 模型为 NioSocketChannel，表示 IO 模型为 NIO。
3. 接着，给引导类指定一个 handler，这里主要就是定义连接的业务处理逻辑。
4. 配置完线程模型、IO 模型、业务处理逻辑之后，调用 connect 方法进行连接，可以看到 connect 方法有两个参数，第一个参数可以填写 IP 或者域名，第二个参数填写的是端口号，由于 connect 方法返回的是一个 Future，也就是说这个方法是异步的，我们通过 addListener 方法可以监听到连接是否成功，进而打印出连接信息。

核心组件

EventLoop

在 Netty 中 EventLoop 可以理解为 Reactor 线程模型的事件处理引擎，每个 EventLoop 线程都维护一个 Selector 选择器和任务队列 taskQueue。

它主要负责处理 I/O 事件、普通任务和定时任务。

Netty 中推荐使用 NioEventLoop 作为实现类。

「NioEventLoop 类继承关系」

可以从上面的类继承关系中看到，NioEventLoop继承了大量的接口，实现了大量的数据能力：

1. NioEventLoop是一个单线程的线程池。从这个类继承了SingleThreadEventLoop又实现了ExecutorService就可以知道。
2. EventLoop是可以提交任务的。因为他是一个线程池，所以完全可以提交任务。确切的来说，EventLoop是维护了一个任务队列的。
3. NioEventLoop内部持有的线程，声明周期内。在ThreadPoolExecutor里面，持有线程的方式是通过Worker这种内部类的方式持有一个Thread，在NioEventLoop里面，直接持有了一个Thread。
4. NioEventLoop内部持有一个Selector, 因为NioEventLoop需要执行IO任务。

「看下 EventLoop 的事件流转图：」

BossEventLoopGroup 和 WorkerEventLoopGroup 包含一个或者多个 NioEventLoop。

BossEventLoopGroup 负责监听客户端的 Accept 事件，当事件触发时，将事件注册至 WorkerEventLoopGroup 中的一个 NioEventLoop 上。

每新建一个 Channel， 只选择一个 NioEventLoop 与其绑定。

NioEventLoop 完成数据读取后，会调用绑定的 ChannelPipeline 进行事件传播，ChannelPipeline 也是线程安全的，数据会被传递到 ChannelPipeline 的第一个 ChannelHandler 中。

数据处理完成后，将加工完成的数据再传递给下一个 ChannelHandler，整个过程是串行化执行，不会发生线程上下文切换的问题。

「EventLoopGroup」

可以简单理解为一个EventLoop线程池。

Pipeline

ChannelPipeline 作为 Netty 的核心编排组件，负责调度各种类型的 ChannelHandler，实际数据的加工处理操作则是由 ChannelHandler 完成的。

ChannelPipeline 可以看作是 ChannelHandler 的容器载体，它是由一组 ChannelHandler 实例组成的，内部通过双向链表将不同的 ChannelHandler 链接在一起。

当有 I/O 读写事件触发时，ChannelPipeline 会依次调用 ChannelHandler 列表对 Channel 的数据进行拦截和处理。

每个 Channel 会绑定一个 ChannelPipeline，每一个 ChannelPipeline 都包含多个 ChannelHandlerContext，所有 ChannelHandlerContext 之间组成了双向链表。

根据网络数据的流向，ChannelPipeline 分为入站 ChannelInboundHandler 和出站 ChannelOutboundHandler 两种处理器。

在客户端与服务端通信的过程中，数据从客户端发向服务端的过程叫出站，反之称为入站。

数据先由一系列 InboundHandler 处理后入站，然后再由相反方向的 OutboundHandler 处理完成后出站。

我们经常使用的解码器 Decoder 就是入站操作，编码器 Encoder 就是出站操作。

服务端接收到客户端数据需要先经过 Decoder 入站处理后，再通过 Encoder 出站通知客户端。

ChannelHandler

「ChannelHandler的分类」

第一个子接口是 ChannelInboundHandler，他是处理读数据的逻辑，比如，我们在一端读到一段数据，首先要解析这段数据，然后对这些数据做一系列逻辑处理，最终把响应写到对端， 在开始组装响应之前的所有的逻辑，都可以放置在 ChannelInboundHandler 里处理，它的一个最重要的方法就是 channelRead()。

第二个子接口 ChannelOutBoundHandler 是处理写数据的逻辑，它是定义我们一端在组装完响应之后，把数据写到对端的逻辑，比如，我们封装好一个 response 对象，接下来我们有可能对这个 response 做一些其他的特殊逻辑，然后，再编码成 ByteBuf，最终写到对端，它里面最核心的一个方法就是 write()。

这两个子接口分别有对应的默认实现，ChannelInboundHandlerAdapter，和 ChanneloutBoundHandlerAdapter，它们分别实现了两大接口的所有功能，默认情况下会把读写事件传播到下一个 handler。

「ChannelInboundHandler的事件传播」

我们在服务端的 pipeline 添加三个 ChannelInboundHandler

1. serverBootstrap 
2.         .childHandler(new ChannelInitializer<NioSocketChannel>() { 
3.             protected void initChannel(NioSocketChannel ch) { 
4.                 ch.pipeline().addLast(new InBoundHandlerA()); 
5.                 ch.pipeline().addLast(new InBoundHandlerB()); 
6.                 ch.pipeline().addLast(new InBoundHandlerC()); 
7.             } 
8.         }); 

每个 inBoundHandler 都继承自 ChannelInboundHandlerAdapter，然后实现了 channelRead() 方法

1. public class InBoundHandlerA extends ChannelInboundHandlerAdapter { 
2.     @Override 
3.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
4.         System.out.println("InBoundHandlerA: " + msg); 
5.         super.channelRead(ctx, msg); 
6.     } 
7. } 
8.  
9. public class InBoundHandlerB extends ChannelInboundHandlerAdapter { 
10.     @Override 
11.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
12.         System.out.println("InBoundHandlerB: " + msg); 
13.         super.channelRead(ctx, msg); 
14.     } 
15. } 
16.  
17. public class InBoundHandlerC extends ChannelInboundHandlerAdapter { 
18.     @Override 
19.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
20.         System.out.println("InBoundHandlerC: " + msg); 
21.         super.channelRead(ctx, msg); 
22.     } 
23. } 

在 channelRead() 方法里面，我们打印当前 handler 的信息，然后调用父类的 channelRead() 方法，而这里父类的 channelRead() 方法会自动调用到下一个 inBoundHandler 的 channelRead() 方法，并且会把当前 inBoundHandler 里处理完毕的对象传递到下一个 inBoundHandler，我们例子中传递的对象都是同一个 msg。

「ChannelOutboundHandler的事件传播」

我们继续在服务端的 pipeline 添加三个 ChanneloutBoundHandler

1. serverBootstrap 
2.         .childHandler(new ChannelInitializer<NioSocketChannel>() { 
3.             protected void initChannel(NioSocketChannel ch) { 
4.                 // inBound，处理读数据的逻辑链 
5.                 ch.pipeline().addLast(new InBoundHandlerA()); 
6.                 ch.pipeline().addLast(new InBoundHandlerB()); 
7.                 ch.pipeline().addLast(new InBoundHandlerC()); 
8.                  
9.                 // outBound，处理写数据的逻辑链 
10.                 ch.pipeline().addLast(new OutBoundHandlerA()); 
11.                 ch.pipeline().addLast(new OutBoundHandlerB()); 
12.                 ch.pipeline().addLast(new OutBoundHandlerC()); 
13.             } 
14.         }); 

每个 outBoundHandler 都继承自 ChanneloutBoundHandlerAdapter，然后实现了 write() 方法

1. public class OutBoundHandlerA extends ChannelOutboundHandlerAdapter { 
2.     @Override 
3.     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
4.         System.out.println("OutBoundHandlerA: " + msg); 
5.         super.write(ctx, msg, promise); 
6.     } 
7. } 
8.  
9. public class OutBoundHandlerB extends ChannelOutboundHandlerAdapter { 
10.     @Override 
11.     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
12.         System.out.println("OutBoundHandlerB: " + msg); 
13.         super.write(ctx, msg, promise); 
14.     } 
15. } 
16.  
17. public class OutBoundHandlerC extends ChannelOutboundHandlerAdapter { 
18.     public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { 
19.         System.out.println("OutBoundHandlerC: " + msg); 
20.         super.write(ctx, msg, promise); 
21.     } 
22. } 

在 write() 方法里面，我们打印当前 handler 的信息，然后调用父类的 write() 方法，而这里父类的 write() 方法会自动调用到下一个 outBoundHandler 的 write() 方法，并且会把当前 outBoundHandler 里处理完毕的对象传递到下一个 outBoundHandler。

不管我们定义的是哪种类型的 handler, 最终它们都是以双向链表的方式连接，这里实际链表的节点是 ChannelHandlerContext。

「ChannelHandler的生命周期」

ChannelHandler 的回调方法，这些回调方法的执行是有顺序的，而这个执行顺序可以称为 ChannelHandler 的生命周期。

ChannelHandler 回调方法的执行顺序为：

• handlerAdded() -> channelRegistered() -> channelActive() -> channelRead() -> channelReadComplete()
• channelInactive() -> channelUnregistered() -> handlerRemoved()

「每个回调方法的含义：」

• handlerAdded() ：指的是当检测到新连接之后，调用 ch.pipeline().addLast(new LifeCyCleTestHandler());之后的回调，表示在当前的 channel 中，已经成功添加了一个 handler 处理器。
• channelRegistered()：这个回调方法，表示当前的 channel 的所有的逻辑处理已经和某个 NIO 线程建立了绑定关系。
• channelActive()：当 channel 的所有的业务逻辑链准备完毕(也就是说 channel 的 pipeline 中已经添加完所有的 handler)以及绑定好一个 NIO 线程之后，这条连接算是真正激活了，接下来就会回调到此方法。
• channelRead()：客户端向服务端发来数据，每次都会回调此方法，表示有数据可读。
• channelReadComplete()：服务端每次读完一次完整的数据之后，回调该方法，表示数据读取完毕。
• channelInactive(): 表面这条连接已经被关闭了，这条连接在 TCP 层面已经不再是 ESTABLISH 状态了。
• channelUnregistered(): 既然连接已经被关闭，那么与这条连接绑定的线程就不需要对这条连接负责了，这个回调就表明与这条连接对应的 NIO 线程移除掉对这条连接的处理。
• handlerRemoved()：最后，我们给这条连接上添加的所有的业务逻辑处理器都给移除掉。

ByteBuf

ByteBuf 是 Netty 的数据容器，所有网络通信中字节流的传输都是通过 ByteBuf 完成的。

「ByteBuf 的结构：」

从上面这幅图可以看到：

1. ByteBuf 是一个字节容器，容器里面的的数据分为三个部分，第一个部分是已经丢弃的字节，这部分数据是无效的;
2. 第二部分是可读字节，这部分数据是 ByteBuf 的主体数据，从 ByteBuf 里面读取的数据都来自这一部分;
3. 最后一部分的数据是可写字节，所有写到 ByteBuf 的数据都会写到这一段。
4. 最后一部分虚线表示的是该 ByteBuf 最多还能扩容多少容量

从 ByteBuf 中每读取一个字节，readerIndex 自增1，ByteBuf 里面总共有 writerIndex-readerIndex 个字节可读，由此可以推论出当 readerIndex 与 writerIndex 相等的时候，ByteBuf 不可读。

写数据是从 writerIndex 指向的部分开始写，每写一个字节，writerIndex 自增1，直到增到 capacity，这个时候，表示 ByteBuf 已经不可写了。

ByteBuf 里面其实还有一个参数 maxCapacity，当向 ByteBuf 写数据的时候，如果容量不足，那么这个时候可以进行扩容，直到 capacity 扩容到 maxCapacity，超过 maxCapacity 就会报错。

「ByteBuffer类的操作」

• 分配缓冲区

ByteBuf 分配一个缓冲区，仅仅给定一个初始值就可以。

1. ByteBuf buf = Unpooled.buffer(13); 
2. System.out.println(String.format("init: ridx=%s widx=%s cap=%s", buf.readerIndex(), buf.writerIndex(), buf.capacity())); 

• 写操作

ByteBuf 写操作和 ByteBuffer 类似，只是写指针是单独记录的，ByteBuf 的写操作支持多种类型。

写入字节数组类型：

1. String content = "月伴飞鱼公众号"; 
2. buf.writeBytes(content.getBytes()); 
3. System.out.println(String.format("write: ridx=%s widx=%s cap=%s", buf.readerIndex(), buf.writerIndex(), buf.capacity())); 

• 读操作

一样的，ByteBuf 写操作和 ByteBuffer 类似，只是写指针是单独记录的，ByteBuf 的读操作支持多种类型。

从当前 readerIndex 位置读取四个字节内容：

1. byte[] dst = new byte[4]; 
2. buf.readBytes(dst); 
3. System.out.println(new String(dst)); 
4. System.out.println(String.format("read(4): ridx=%s widx=%s cap=%s", buf.readerIndex(), buf.writerIndex(), buf.capacity())); 

编解码

每个网络应用程序都必须定义如何解析在两个节点之间来回传输的原始字节，以及如何将其和目标应用程序的数据格式做相互转换。

这种转换逻辑由编解码器处理，编解码器由编码器和解码器组成，它们每种都可以将字节流从一种格式转换为另一种格式。

编码器是将消息转换为适合于传输的格式(最有可能的就是字节流);而对应的解码器则是将网络字节流转换回应用程序的消息格式。

• 解码器：负责处理入站InboundHandler数据，将字节数组转换为消息对象
• 编码器：负责处理出站OutboundHandler数据，将消息对象转换为字节数组

当通过Netty发送或者接受一个消息的时候，就会发生一次数据的转换。入站消息会被解码，出站消息会被编码。

Netty提供了一系列实用的编码解码器，他们都实现了ChannelInboundHadnler或者ChannelOutboundHandler接口。

在这些类中，channelRead方法已经被重写了。

以入站为例，对于每个从入站Channel读取的消息，这个方法会被调用。

随后，它将调用由已知解码器所提供的decode()方法进行解码，并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。

解码器

Netty作为网络框架，提供了大部分目前技术应用非常常见的解码器，提供开箱即使用的功能，所有解码器可以看一下io.netty.handler.codec包中的各种实现

因为解码器是负责将入站数据从一种格式转换到另一种格式的，所以Netty 的解码器其实也实现了ChannelInboundHandler 。

「什么时候会用到解码器?」

每当需要为ChannelPipeline 中的下一个ChannelInboundHandler 转换入站数据时会用到。

「自定义编解码器」

• 通过继承ByteToMessageDecoder自定义解码器。
• 通过继承MessageToByteEncoder自定义编码器。

「ByteToMessageDecoder」

核心方法-decode()方法被调用时将会传入一个包含了传入数据的ByteBuf，以及一个用来添加解码消息的List。

对这个方法的调用将会重复进行，直到确定没有新的元素被添加到该List，或者该ByteBuf 中没有更多可读取的字节时为止。

然后，如果该List 不为空，那么它的内容将会被传递给ChannelPipeline 中的下一个ChannelInboundHandler。

1. protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception; 

「ToIntegerDecoder」

举了一个简单例子，实现了一个Integer的解码器。

假设接收了一个包含简单int 的字节流，每个int都需要被单独处理。

在这种情况下，需要从入站ByteBuf 中读取每个int，并将它传递给ChannelPipeline 中的下一个ChannelInboundHandler。

为了解码这个字节流，可以扩展ByteToMessageDecoder 类来完成。

1. public class ToIntegerDecoder extends ByteToMessageDecoder { 
2.   @Override 
3.   public void decode(ChannelHandlerContext ctx, ByteBuf in,List<Object> out) throws Exception { 
4.     if (in.readableBytes() >= 4) { 
5.     out.add(in.readInt()); 
6.     } 
7.   } 
8. } 

「ReplayingDecoder」

ReplayingDecoder扩展了ByteToMessageDecoder类，使得我们不必调用readableBytes()方法。

它通过使用一个自定义的ByteBuf实现，ReplayingDecoderByteBuf，包装传入的ByteBuf实现了这一点，其将在内部执行该调用

1. public class ToIntegerDecoder2 extends ReplayingDecoder<Void> { 
2.    @Override 
3.    public void decode(ChannelHandlerContext ctx, ByteBuf in,List<Object> out) throws Exception { 
4.      out.add(in.readInt()); 
5.    } 
6. } 

「MessageToMessageDecoder」

该类就如名字一样在两个消息格式之间进行转换，例如，从一种POJO 类型转换为另一种

1. protected abstract void decode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception; 

「IntegerToStringDecoder」

一个例子，将integer转换为String的Decoder

1. public class IntegerToStringDecoder extends MessageToMessageDecoder<Integer> { 
2.   @Override 
3.   public void decode(ChannelHandlerContext ctx, Integer msg,List<Object> out) throws Exception { 
4.     out.add(String.valueOf(msg)); 
5.   } 
6. } 

编码器

编码器其实就是解码器的逆向过程，在Netty当中，只需要把要写入网络的字节写入到ByteBuf当中即可

1. protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception; 

拆包粘包

「为什么有拆包/粘包?」

TCP 传输协议是面向流的，没有数据包界限。客户端向服务端发送数据时，可能将一个完整的报文拆分成多个小报文进行发送，也可能将多个报文合并成一个大的报文进行发送，因此就有了拆包和粘包。

我们需要知道，尽管我们在应用层面使用了 Netty，但是对于操作系统来说，只认 TCP 协议，尽管我们的应用层是按照 ByteBuf 为 单位来发送数据，但是到了底层操作系统仍然是按照字节流发送数据，因此，数据到了服务端，也是按照字节流的方式读入，然后到了 Netty 应用层面，重新拼装成 ByteBuf，而这里的 ByteBuf 与客户端按顺序发送的 ByteBuf 可能是不对等的。

因此，我们需要在客户端根据自定义协议来组装我们应用层的数据包，然后在服务端根据我们的应用层的协议来组装数据包，这个过程通常在服务端称为拆包，而在客户端称为粘包。

「粘包和拆包的解决方法」

由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界的主流协议的解决方案，可以归纳如下：

消息长度固定，累计读取到长度和为定长LEN的报文后，就认为读取到了一个完整的信息

将回车换行符作为消息结束符

将特殊的分隔符作为消息的结束标志，回车换行符就是一种特殊的结束分隔符

通过在消息头中定义长度字段来标识消息的总长度

「Netty中的粘包和拆包解决方案」

Netty提供了4种解码器来解决，分别如下：

• FixedLengthFrameDecoder-定长协议解码器，我们可以指定固定的字节数算一个完整的报文
• LineBasedFrameDecoder-行分隔符解码器，遇到\n或者\r\n，则认为是一个完整的报文
• DelimiterBasedFrameDecoder-自定义分隔符解码器，与LineBasedFrameDecoder类似，只不过分隔符可以自己指定
• LengthFieldBasedFrameDecoder-自定义协议头解码器，将报文划分为报文头/报文体，根据报文头中的Length字段确定报文体的长度，因此报文体的长度是可变的

「LineBasedFrameDecoder解码器」

1. private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { 
2.  
3.         @Override 
4.         protected void initChannel(SocketChannel socketChannel) throws Exception { 
5.             // 管道（Pipeline）持有某个通道的全部处理器 
6.             ChannelPipeline pipeline = socketChannel.pipeline(); 
7.             // 解决粘包问题 
8.             pipeline.addLast(new LineBasedFrameDecoder(1024)); 
9.             pipeline.addLast(new StringDecoder()); 
10.             // 添加处理器 
11.             pipeline.addLast(new NettyServerHandler()); 
12.  
13.         } 
14.     } 

当然我们在 客户端发送消息的时候就需要加上分割符号(这边以换行符号作为分割)，要不然服务端没法接收消息

1. @Override 
2.   public void channelActive(ChannelHandlerContext ctx) throws Exception { 
3.       for (int i=0;i<100;i++){ 
4.           byte[] bytes = "关注公众号月伴飞鱼\n".getBytes(); 
5.           // 创建节字缓冲区 
6.           ByteBuf message = Unpooled.buffer(bytes.length); 
7.           // 将数据写入缓冲区 
8.           message.writeBytes(bytes); 
9.           // 写入数据 
10.           ctx.writeAndFlush(message); 
11.       } 
12.  
13.   } 

LineBasedFrameDecoder 支持 \n 或者 \n\r 进行解码;

「DelimiterBasedFrameDecoder 解码器」

DelimiterBasedFrameDecoder 解码器的应用和 LineBasedFrameDecoder 相似，不过 优点是我们可以自定义分割符号;

1. private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { 
2.  
3.         @Override 
4.         protected void initChannel(SocketChannel socketChannel) throws Exception { 
5.             // 管道（Pipeline）持有某个通道的全部处理器 
6.             ChannelPipeline pipeline = socketChannel.pipeline(); 
7.             pipeline.addLast(new DelimiterBasedFrameDecoder(10240, Unpooled.copiedBuffer("%".getBytes()))); 
8.             pipeline.addLast(new StringDecoder()); 
9.             // 添加处理器 
10.             pipeline.addLast(new NettyServerHandler()); 
11.  
12.         } 
13.     } 

我们在客户端发送数据的时候以 % 为结尾;

1. @Override 
2.    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
3.        for (int i=0;i<100;i++){ 
4.            
5.            byte[] bytes = "关注公众号月伴飞鱼%".getBytes(); 
6.            // 创建节字缓冲区 
7.            ByteBuf message = Unpooled.buffer(bytes.length); 
8.            // 将数据写入缓冲区 
9.            message.writeBytes(bytes); 
10.            // 写入数据 
11.            ctx.writeAndFlush(message); 
12.        } 
13.  
14.    } 

「FixedLengthFrameDecoder 解码器」

固定长度解码器 FixedLengthFrameDecoder 非常简单，直接通过构造函数设置固定长度的大小 frameLength，无论接收方一次获取多大的数据，都会严格按照 frameLength 进行解码。

如果累积读取到长度大小为 frameLength 的消息，那么解码器认为已经获取到了一个完整的消息。

如果消息长度小于 frameLength，FixedLengthFrameDecoder 解码器会一直等后续数据包的到达，直至获得完整的消息。

1. private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { 
2.  
3.         @Override 
4.         protected void initChannel(SocketChannel socketChannel) throws Exception { 
5.             // 管道（Pipeline）持有某个通道的全部处理器 
6.             ChannelPipeline pipeline = socketChannel.pipeline(); 
7.             pipeline.addLast(new FixedLengthFrameDecoder(63)); 
8.             pipeline.addLast(new StringDecoder()); 
9.             // 添加处理器 
10.             pipeline.addLast(new NettyServerHandler()); 
11.  
12.         } 
13.     } 

空闲检测

「服务端空闲检测」

对于服务端来说，客户端的连接如果出现假死，那么服务端将无法收到客户端的数据，也就是说，如果能一直收到客户端发来的数据，那么可以说明这条连接还是活的，因此，服务端对于连接假死的应对策略就是空闲检测。

空闲检测指的是每隔一段时间，检测这段时间内是否有数据读写，我们的服务端只需要检测一段时间内，是否收到过客户端发来的数据即可，Netty 自带的 IdleStateHandler 就可以实现这个功能。

接下来，我们写一个类继承自 IdleStateHandler，来定义检测到假死连接之后的逻辑。

1. public class IMIdleStateHandler extends IdleStateHandler { 
2.  
3.     private static final int READER_IDLE_TIME = 15; 
4.  
5.     public IMIdleStateHandler() { 
6.         super(READER_IDLE_TIME, 0, 0, TimeUnit.SECONDS); 
7.     } 
8.  
9.     @Override 
10.     protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) { 
11.         System.out.println(READER_IDLE_TIME + "秒内未读到数据，关闭连接"); 
12.         ctx.channel().close(); 
13.     } 
14. } 

1. 我们看下IMIdleStateHandler 的构造函数，他调用父类 IdleStateHandler 的构造函数，有四个参数，其中第一个表示读空闲时间，指的是在这段时间内如果没有数据读到，就表示连接假死;第二个是写空闲时间，指的是 在这段时间如果没有写数据，就表示连接假死;第三个参数是读写空闲时间，表示在这段时间内如果没有产生数据读或者写，就表示连接假死，写空闲和读写空闲为0，表示我们不关心者两类条件;最后一个参数表示时间单位。
2. 连接假死之后会回调 channelIdle() 方法，我们这个方法里面打印消息，并手动关闭连接。

Recycler

对象池对象池与内存池的都是为了提高 Netty 的并发处理能力，我们知道 Java 中频繁地创建和销毁对象的开销是很大的，所以很多人会将一些通用对象缓存起来，当需要某个对象时，优先从对象池中获取对象实例。

通过重用对象，不仅避免频繁地创建和销毁所带来的性能损耗，而且对 JVM GC 是友好的，这就是对象池的作用。

Recycler 是 Netty 提供的自定义实现的轻量级对象回收站，借助 Recycler 可以完成对象的获取和回收。

「一个示例：」

1. public class RecyclerDemo { 
2.     private static final Recycler<User> RECYCLER = new Recycler<User>() { 
3.         @Override 
4.         protected User newObject(Handle<User> handle) { 
5.             return new User(handle); 
6.         } 
7.     }; 
8.     static class User{ 
9.         private final Recycler.Handle<User> handle; 
10.         public User(Recycler.Handle<User> handle){ 
11.             this.handle=handle; 
12.         } 
13.         public void recycle(){ 
14.             handle.recycle(this); 
15.         } 
16.     } 
17.     public static void main(String[] args){ 
18.         User user1 = RECYCLER.get(); 
19.         user1.recycle(); 
20.         User user2 = RECYCLER.get(); 
21.         user2.recycle(); 
22.         System.out.println(user1==user2); 
23.     } 
24. } 

首先定义了一个Recycler的成员变量RECYCLER，在匿名内部类中重写了newObject方法，也就是创建对象的方法，该方法就是用户自定义的，这里newObject返回的new User(handle)，代表当回收站没有此类对象的时候，可以通过这种方式创建对象

「成员变量RECYCLER, 可以用来对此类对象的回收和再利用」

定义了一个一个静态内部类User，User中有个成员变量handle，在构造方法中为其赋值，handle的作用就是用于对象回收的，并且定义了一个方法recycle，方法体中通过handle.recycle(this)这种方式将自身对象进行回收，通过这步操作就可以将对象回收到Recycler中，在main方法中，通过RECYCLER的get方法获取一个user，然后进行回收，再通过get方法将回收站的对象取出，再次进行回收，最后判断两次取出的对象是否为一个对象， 最后结果输出为true

基本结构

「Recycler 的内部结构：」

通过 Recycler 的 UML 图可以看出，一共包含四个核心组件：Stack、WeakOrderQueue、Link、DefaultHandle。

「核心组件Stack」

Stack 是整个对象池的顶层数据结构，描述了整个对象池的构造，用于存储当前本线程回收的对象。

在多线程的场景下，Netty 为了避免锁竞争问题，每个线程都会持有各自的对象池，内部通过 FastThreadLocal 来实现每个线程的私有化。

「在Recycler的类的源码中, 我们看到这一段逻辑」:

1. private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() { 
2.     @Override 
3.     protected Stack<T> initialValue() { 
4.         return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,  
5.                 ratioMask, maxDelayedQueuesPerThread); 
6.     } 
7. }; 

「Stack类的源码定义：」

每个stack中维护着一个DefaultHandle类型的数组，用于放回收的对象，有关stack和线程的关系如图所示:

「Stack构造方法」:

「几个构造方法中初始化的关键属性:」

• 属性parent表示Reclycer对象自身
• 属性thread表示当前stack绑定的哪个线程
• 属性maxCapacity表示当前stack的最大容量，表示stack最多能盛放多少个元素
• 属性elements, 就表示stack中存储的对象，类型为DefaultHandle，可以被外部对象引用，从而实现回收
• 属性ratioMask是用来控制对象回收的频率的，也就是说每次通过Reclycer回收对象的时候，不是每次都会进行回收，而是通过该参数控制回收频率
• 属性maxDelayedQueues，在很多时候, 一个线程创建的对象，有可能会被另一个线程所释放，而另一个线程释放的对象是不会放在当前线程的stack中的，而是会存放在一个叫做WeakOrderQueue的数据结构中，里面也是存放着一个个DefaultHandle，WeakOrderQueue会存放线程1创建的并且在线程2进行释放的对象

maxDelayedQueues属性的意思就是我这个线程能回收几个其他创建的对象的线程，假设当前线程是线程1， maxDelayedQueues为2，那么我线程1回收了线程2创建的对象，又回收了线程3创建的对象，那么不可能回收线程4创建的对象了，因为maxDelayedQueues为2，只能回收两个线程创建的对象

• 属性availableSharedCapacity, 表示在线程1中创建的对象，在其他线程中缓存的最大个数

「另外介绍两个没有在构造方法初始化的属性:」

1. private WeakOrderQueue cursor, prev; 
2. private volatile WeakOrderQueue head; 

这里相当于指针，用于指向WeakOrderQueue的

核心源码

「从 Recycler 中获取对象」

从对象池中获取对象的入口是在 Recycler#get() 方法：

通过 FastThreadLocal 获取当前线程的唯一栈缓存 Stack，然后尝试从栈顶弹出 DefaultHandle 对象实例，如果 Stack 中没有可用的 DefaultHandle 对象实例，那么会调用 newObject 生成一个新的对象，完成 handle 与用户对象和 Stack 的绑定。

stack.pop()的源码：

如果 Stack 的 elements 数组中有可用的对象实例，直接将对象实例弹出;

如果 elements 数组中没有可用的对象实例，会调用 scavenge 方法，scavenge 的作用是从其他线程回收的对象实例中转移一些到 elements 数组当中，也就是说，它会想办法从 WeakOrderQueue 链表中迁移部分对象实例。

每个 Stack 会有一个 WeakOrderQueue 链表，每个 WeakOrderQueue 节点都维持了相应异线程回收的对象，那么以什么样的策略从 WeakOrderQueue 链表中迁移对象实例呢?

「scavenge 的源码：」

scavenge 的源码中首先会从 cursor 指针指向的 WeakOrderQueue 节点回收部分对象到 Stack 的 elements 数组中，如果没有回收到数据就会将 cursor 指针移到下一个 WeakOrderQueue，重复执行以上过程直至回到到对象实例为止。

「Recycler 对象回收原理」

DefaultHandle#recycle()

从源码中可以看出，在回收对象时，会向 Stack 中 push 对象，push 会分为同线程回收和异线程回收两种情况，分别对应 pushNow 和 pushLater 两个方法，我们逐一进行分析。

「同线程对象回收」

如果是当前线程回收自己分配的对象时，会调用 pushNow 方法：

「异线程对象回收」

异线程回收对象时，并不会添加到 Stack 中，而是会与 WeakOrderQueue 直接打交道，pushLater 的源码：

零拷贝

「传统意义的拷贝」

是在发送数据的时候，传统的实现方式是：

1. File.read(bytes)
2. Socket.send(bytes)

这种方式需要四次数据拷贝和四次上下文切换：

1. 数据从磁盘读取到内核的read buffer
2. 数据从内核缓冲区拷贝到用户缓冲区
3. 数据从用户缓冲区拷贝到内核的socket buffer
4. 数据从内核的socket buffer拷贝到网卡接口(硬件)的缓冲区

「零拷贝的概念」

明显上面的第二步和第三步是没有必要的，通过java的FileChannel.transferTo方法，可以避免上面两次多余的拷贝(当然这需要底层操作系统支持)

1. 调用transferTo，数据从文件由DMA引擎拷贝到内核read buffer
2. 接着DMA从内核read buffer将数据拷贝到网卡接口buffer

上面的两次操作都不需要CPU参与，所以就达到了零拷贝。

JDK NIO中的的transferTo() 方法这个实现依赖于操作系统底层的sendFile()实现的：

1. #include <sys/socket.h> 
2. ssize_t sendfile(int out_fd, int in_fd, off_t *offset, size_t count); 

「Netty中的零拷贝」

主要体现在三个方面：

「1、ByteBuffer」

Netty发送和接收消息主要使用ByteBuffer，ByteBuffer使用堆外内存(DirectMemory)直接进行Socket读写。

如果使用传统的堆内存进行Socket读写，JVM会将堆内存buffer拷贝一份到直接内存中然后再写入socket，多了一次缓冲区的内存拷贝，DirectMemory中可以直接通过DMA发送到网卡接口

「2、Composite Buffers」

传统的ByteBuffer，如果需要将两个ByteBuffer中的数据组合到一起，我们需要首先创建一个size=size1+size2大小的新的数组，然后将两个数组中的数据拷贝到新的数组中。

但是使用Netty提供的组合ByteBuf，就可以避免这样的操作，因为CompositeByteBuf并没有真正将多个Buffer组合起来，而是保存了它们的引用，从而避免了数据的拷贝，实现了零拷贝。

「3、对于FileChannel.transferTo的使用」

Netty中使用了FileChannel的transferTo方法，该方法依赖于操作系统实现零拷贝。

CompositeByteBuf

CompositeByteBuf 是 Netty 中实现零拷贝机制非常重要的一个数据结构，CompositeByteBuf 可以理解为一个虚拟的 Buffer 对象，它是由多个 ByteBuf 组合而成，但是在 CompositeByteBuf 内部保存着每个 ByteBuf 的引用关系，从逻辑上构成一个整体。

比较常见的像 HTTP 协议数据可以分为头部信息 header和消息体数据 body，分别存在两个不同的 ByteBuf 中，通常我们需要将两个 ByteBuf 合并成一个完整的协议数据进行发送，可以使用如下方式完成：

1. ByteBuf httpBuf = Unpooled.buffer(header.readableBytes() + body.readableBytes()); 
2. httpBuf.writeBytes(header); 
3. httpBuf.writeBytes(body); 

可以看出，如果想实现 header 和 body 这两个 ByteBuf 的合并，需要先初始化一个新的 httpBuf，然后再将 header 和 body 分别拷贝到新的 httpBuf。

合并过程中涉及两次 CPU 拷贝，这非常浪费性能。

如果使用 CompositeByteBuf 如何实现类似的需求：

1. CompositeByteBuf httpBuf = Unpooled.compositeBuffer(); 
2. httpBuf.addComponents(true, header, body); 

CompositeByteBuf 通过调用 addComponents() 方法来添加多个 ByteBuf，但是底层的 byte 数组是复用的，不会发生内存拷贝。

但对于用户来说，它可以当作一个整体进行操作。

「CompositeByteBuf 的内部结构：」

从图上可以看出，CompositeByteBuf 内部维护了一个 Components 数组。

在每个 Component 中存放着不同的 ByteBuf，各个 ByteBuf 独立维护自己的读写索引，而 CompositeByteBuf 自身也会单独维护一个读写索引。

「文件传输 FileRegion」

Netty 使用 FileRegion 实现文件传输的零拷贝。

FileRegion 的默认实现类是 DefaultFileRegion，通过 DefaultFileRegion 将文件内容写入到 NioSocketChannel。

从源码可以看出，FileRegion 其实就是对 FileChannel 的包装，并没有什么特殊操作，底层使用的是 JDK NIO 中的 FileChannel#transferTo() 方法实现文件传输，所以 FileRegion 是操作系统级别的零拷贝，对于传输大文件会很有帮助。

Reactor线程模型

Reactor是反应堆的意思，Reactor模型，是指通过一个或多个输入同时传递给服务处理器的服务请求的事件驱动处理模式。

服务端程序处理传入多路请求，并将它们同步分派给请求对应的处理线程，Reactor模式也叫Dispatcher模式，即I/O多了复用统一监听事件，收到事件后分发(Dispatch给某进程)。

「Reactor 的线程模型有三种:」

• 单线程模型
• 多线程模型
• 主从多线程模型

单线程模型

所谓单线程，即 acceptor 处理和 handler 处理都在一个线程中处理

当其中某个 handler 阻塞时，会导致其他所有的 client 的 handler 都得不到执行，并且更严重的是，handler 的阻塞也会导致整个服务不能接收新的 client 请求(因为 acceptor 也被阻塞了)

多线程模型

Reactor 的多线程模型与单线程模型的区别就是 acceptor 是一个单独的线程处理，并且有一组特定的 NIO 线程来负责各个客户端连接的 IO 操作

Reactor 多线程模型 有如下特点:

• 有专门一个线程，即 Acceptor 线程用于监听客户端的TCP连接请求。
• 客户端连接的 IO 操作都是由一个特定的 NIO 线程池负责，每个客户端连接都与一个特定的 NIO 线程绑定，因此在这个客户端连接中的所有 IO 操作都是在同一个线程中完成的。
• 客户端连接有很多，但是 NIO 线程数是比较少的。因此一个 NIO 线程可以同时绑定到多个客户端连接中。

主从多线程模型

如果我们的服务器需要同时处理大量的客户端连接请求或我们需要在客户端连接时，进行一些权限的检查，那么单线程的 Acceptor 很有可能就处理不过来，造成了大量的客户端不能连接到服务器。

Reactor 的主从多线程模型就是在这样的情况下提出来的，它的特点是: 服务器端接收客户端的连接请求不再是一个线程，而是由一个独立的线程池组成

这种模型在许多项目中广泛使用，包括 Nginx 主从 Reactor 多进程模型， Memcached 主从多线程，Netty 主从多线程模型的支持

「3种模式用生活案例来理解：」

1. 单 Reactor 单线程，前台接待员和服务员是同一个人，全程为顾客服务
2. 单 Reactor 多线程，1 个前台接待员，多个服务员，接待员只负责接待
3. 主从 Reactor 多线程，多个前台接待员，多个服务生

Netty架构模型Netty是一个异步网络通信框架，异步主要体现在对java Future的拓展，基于Future/Listener的回调机制完成了对事件的监听

通过Channel完成了对数据的传输，使用NioEventLoop工作线程，通过执行ChanelHandler完成了对ChannelPipeline上的数据处理。

Future回调机制

Java的Future大家应该比较清楚，以一种非阻塞的方式，快速返回

但是这种方式一个比较大的缺点是用户必须通过.get()方式来获取结果，无法精确了解完成时间。

Netty扩展了Java的Future，最主要的改进就是增加了监听器Listener接口，通过监听器可以让异步执行更加有效率，不需要通过get来等待异步执行结束，而是通过监听器回调来精确地控制异步执行结束的时间点。

ChannelFuture接口扩展了Netty的Future接口，表示一种没有返回值的异步调用，同时关联了Channel，跟一个Channel绑定。

「常见有如下操作：」

• 通过 isDone 方法来判断当前操作是否完成;
• 通过 isSuccess 方法来判断已完成的当前操作是否成功;
• 通过 getCause 方法来获取已完成的当前操作失败的原因;
• 通过 isCancelled 方法来判断已完成的当前操作是否被取消;
• 通过 addListener 方法来注册监听器，当操作已完成(isDone 方法返回完成)，将会通知 指定的监听器;如果 Future 对象已完成，则通知指定的监听器

代码示例：

1. private void doConnect(final Logger logger,final String host, final int port) { 
2.     ChannelFuture future = bootstrap.connect(new InetSocketAddress(host, port)); 
3.  
4.     future.addListener(new ChannelFutureListener() { 
5.         public void operationComplete(ChannelFuture f) throws Exception { 
6.             if (!f.isSuccess()) { 
7.                 logger.info("Started Tcp Client Failed"); 
8.                 f.channel().eventLoop().schedule( new Runnable() { 
9.                     @Override 
10.                     public void run() { 
11.                         doSomeThing(); 
12.                     } 
13.                 }, 200, TimeUnit.MILLISECONDS); 
14.             } 
15.         } 
16.     }); 
17. } 

FastThreadLocal

我们都有在源码中发现 FastThreadLocal 的身影。

Netty 作为高性能的网络通信框架，FastThreadLocal 是比 JDK 自身的 ThreadLocal 性能更高的通信框架。

FastThreadLocal 的实现与 ThreadLocal 非常类似，Netty 为 FastThreadLocal 量身打造了 FastThreadLocalThread 和 InternalThreadLocalMap 两个重要的类。

FastThreadLocalThread 是对 Thread 类的一层包装，每个线程对应一个 InternalThreadLocalMap 实例。

只有 FastThreadLocal 和 FastThreadLocalThread 组合使用时，才能发挥 FastThreadLocal 的性能优势。

首先看下 FastThreadLocalThread 的源码定义：

1. public class FastThreadLocalThread extends Thread { 
2.     private InternalThreadLocalMap threadLocalMap; 
3.     // 省略其他代码 
4. } 

可以看出 FastThreadLocalThread 主要扩展了 InternalThreadLocalMap 字段，我们可以猜测到 FastThreadLocalThread 主要使用 InternalThreadLocalMap 存储数据，而不再是使用 Thread 中的 ThreadLocalMap。

FastThreadLocal 使用 Object 数组替代了 Entry 数组，Object[0] 存储的是一个Set

InternalThreadLocalMap源码：

InternalThreadLocalMap 并没有使用线性探测法来解决 Hash 冲突，而是在 FastThreadLocal 初始化的时候分配一个数组索引 index，index 的值采用原子类 AtomicInteger 保证顺序递增，通过调用InternalThreadLocalMap.nextVariableIndex()方法获得。

然后在读写数据的时候通过数组下标 index 直接定位到 FastThreadLocal 的位置，时间复杂度为 O(1)。

如果数组下标递增到非常大，那么数组也会比较大，所以 FastThreadLocal 是通过空间换时间的思想提升读写性能。

核心源码

「FastThreadLocal.set() 的源码：」

InternalThreadLocalMap.get()方法，源码如下：

如果当前线程是 FastThreadLocalThread 类型，那么直接通过 fastGet() 方法获取 FastThreadLocalThread 的 threadLocalMap 属性即可。

slowGet() 是针对非 FastThreadLocalThread 类型的线程发起调用时的一种兜底方案。

如果当前线程不是 FastThreadLocalThread，内部是没有 InternalThreadLocalMap 属性的，Netty 在 UnpaddedInternalThreadLocalMap 中保存了一个 JDK 原生的 ThreadLocal，ThreadLocal 中存放着 InternalThreadLocalMap，此时获取 InternalThreadLocalMap 就退化成 JDK 原生的 ThreadLocal 获取。

「FastThreadLocal.get() 的源码实现如下：」

首先根据当前线程是否是 FastThreadLocalThread 类型找到 InternalThreadLocalMap，然后取出从数组下标 index 的元素，如果 index 位置的元素不是缺省对象 UNSET，说明该位置已经填充过数据，直接取出返回即可。

如果 index 位置的元素是缺省对象 UNSET，那么需要执行初始化操作。

原文链接：https://mp.weixin.qq.com/s/NufP2G7TJBQKiSeQmI3-EQ