Netty 源码解析（四）: Netty 的 ChannelPipeline

今天是猿灯塔“365篇原创计划”第四篇。

接下来的时间灯塔君持续更新Netty系列一共九篇

Netty 源码解析（一）: 开始

Netty 源码解析（二）: Netty 的 Channel

Netty 源码解析（三）: Netty 的 Future 和 Promise

当前：Netty 源码解析（四）: Netty 的 ChannelPipeline

Netty 源码解析（五）: Netty 的线程池分析

Netty 源码解析（六）: Channel 的 register 操做

Netty 源码解析（七）: NioEventLoop 工做流程

Netty 源码解析（八）: 回到 Channel 的 register 操做

Netty 源码解析（九）: connect 过程和 bind 过程分析

今天呢！灯塔君跟你们讲：

Netty 的 ChannelPipeline

ChannelPipeline和Inbound、Outbound

我想不少读者应该或多或少都有 Netty 中 pipeline 的概念。前面咱们说了，使用 Netty 的时候，咱们一般就只要写一些自定义的 handler 就能够了，咱们定义的这些 handler 会组成一个 pipeline，用于处理 IO 事件，这个和咱们平时接触的 Filter 或 Interceptor 表达的差很少是一个意思。

每一个 Channel 内部都有一个 pipeline，pipeline 由多个 handler 组成，handler 之间的顺序是很重要的，由于 IO 事件将按照顺序顺次通过 pipeline 上的 handler，这样每一个 handler 能够专一于作一点点小事，由多个 handler 组合来完成一些复杂的逻辑。

从图中，咱们知道这是一个双向链表。

首先，咱们看两个重要的概念：Inbound 和 Outbound。在 Netty 中，IO 事件被分为 Inbound 事件和 Outbound 事件。

Outbound 的 out 指的是 出去，有哪些 IO 事件属于此类呢？好比 connect、write、flush 这些 IO 操做是往外部方向进行的，它们就属于 Outbound 事件。

其余的，诸如 accept、read 这种就属于 Inbound 事件。

好比客户端在发起请求的时候，须要 1️⃣connect 到服务器，而后 2️⃣write 数据传到服务器，再而后 3️⃣read 服务器返回的数据，前面的 connect 和 write 就是 out 事件，后面的 read 就是 in 事件。

好比不少初学者看不懂下面的这段代码，这段代码用于服务端的 childHandler 中：

pipeline.addLast(new StringDecoder());

pipeline.addLast(new StringEncoder());

pipeline.addLast(new BizHandler());

初学者确定都纳闷，觉得这个顺序写错了，应该是先 decode 客户端过来的数据，而后用 BizHandler 处理业务逻辑，最后再 encode 数据而后返回给客户端，因此添加的顺序应该是 1 -> 3 -> 2 才对。

其实这里的三个 handler 是分组的，分为 Inbound（1 和 3） 和 Outbound（2）：

1. pipeline.addLast(new StringDecoder());
2. pipeline.addLast(new StringEncoder());
3. pipeline.addLast(new BizHandler());

• 客户端链接进来的时候，读取（read）客户端请求数据的操做是 Inbound 的，e 操做是 Outbound 的，此时使用的是 2。
• 处理完数据后，返回给客户端数据的 write 操做是 Outbound 的，此时使用的是 2。

因此虽然添加顺序有点怪，可是执行顺序实际上是按照 1 -> 3 -> 2 进行的。

若是咱们在上面的基础上，加上下面的第四行，这是一个 OutboundHandler：

1. pipeline.addLast(new OutboundHandlerA());

那么执行顺序是否是就是 1 -> 3 -> 2 -> 4 呢？答案是：不是的。

对于 Inbound 操做，按照添加顺序执行每一个 Inbound 类型的 handler；而对于 Outbound 操做，是反着来的，从后往前，顺次执行 Outbound 类型的 handler。

因此，上面的顺序应该是先 1 后 3，它们是 Inbound 的，而后是 4，最后才是 2，它们两个是 Outbound 的。说实话，这种组织方式对新手应该非常头疼。

那咱们在开发的时候怎么写呢？其实也很简单，从最外层开始写，一步步写到业务处理层，把 Inbound 和 Outbound 混写在一块儿。好比 encode 和 decode 是属于最外层的处理逻辑，先写它们。假设 decode 之后是字符串，那再进来一层应该能够写进来和出去的日志。再进来一层能够写 字符串 <=> 对象 的相互转换。而后就应该写业务层了。

到这里，我想你们应该都知道 Inbound 和 Outbound 了吧？下面咱们来介绍它们的接口使用。

定义处理 Inbound 事件的 handler 须要实现 ChannelInboundHandler，定义处理 Outbound 事件的 handler 须要实现 ChannelOutboundHandler。最下面的三个类，是 Netty 提供的适配器，特别的，若是咱们但愿定义一个 handler 能同时处理 Inbound 和 Outbound 事件，能够经过继承中间的 ChannelDuplexHandler 的方式，好比 LoggingHandler 这种既能够用来处理 Inbound 也能够用来处理 Outbound 事件的 handler。

有了 Inbound 和 Outbound 的概念之后，咱们来开始介绍 Pipeline 的源码。

咱们说过，一个 Channel 关联一个 pipeline，NioSocketChannel 和 NioServerSocketChannel 在执行构造方法的时候，都会走到它们的父类 AbstractChannel 的构造方法中：

`protected AbstractChannel(Channel parent) {

this.parent = parent;

// 给每一个 channel 分配一个惟一 id

id = newId();

// 每一个 channel 内部须要一个 Unsafe 的实例

unsafe = newUnsafe();

// 每一个 channel 内部都会建立一个 pipeline

pipeline = newChannelPipeline();

}`

上面的三行代码中，id 比较不重要，Netty 中的 Unsafe 实例其实挺重要的，这里简单介绍一下。

在 JDK 的源码中，sun.misc.Unsafe 类提供了一些底层操做的能力，它设计出来是给 JDK 中的源码使用的，好比 AQS、ConcurrentHashMap 等，咱们在以前的并发包的源码分析中也看到了不少它们使用 Unsafe 的场景，这个 Unsafe 类不是给咱们的代码使用的，是给 JDK 源码使用的（须要的话，咱们也是能够获取它的实例的）。

Unsafe 类的构造方法是 private 的，可是它提供了 getUnsafe() 这个静态方法：

Unsafe unsafe = Unsafe.getUnsafe();

你们能够试一下，上面这行代码编译没有问题，可是执行的时候会抛 java.lang.SecurityException 异常，由于它就不是给咱们的代码用的。

可是若是你就是想获取 Unsafe 的实例，能够经过下面这个代码获取到:

Field f = Unsafe.class.getDeclaredField("theUnsafe");

f.setAccessible(true);

Unsafe unsafe = (Unsafe) f.get(null);

Netty 中的 Unsafe 也是一样的意思，它封装了 Netty 中会使用到的 JDK 提供的 NIO 接口，好比将 channel 注册到 selector 上，好比 bind 操做，好比 connect 操做等，这些操做都是稍微偏底层一些。Netty 一样也是不但愿咱们的业务代码使用 Unsafe 的实例，它是提供给 Netty 中的源码使用的。

不过，对于咱们源码分析来讲，咱们仍是会有不少时候须要分析 Unsafe 中的源码的

关于 Unsafe，咱们后面用到了再说，这里只要知道，它封装了大部分须要访问 JDK 的 NIO 接口的操做就行了。这里咱们继续将焦点放在实例化 pipeline 上：

`protected DefaultChannelPipeline newChannelPipeline() {

return new DefaultChannelPipeline(this);

}`

这里开始调用 DefaultChannelPipeline 的构造方法，并把当前 channel 的引用传入：

`protected DefaultChannelPipeline(Channel channel) {
this.channel = ObjectUtil.checkNotNull(channel, "channel");
succeededFuture = new SucceededChannelFuture(channel, null);
voidPromise = new VoidChannelPromise(channel, true);
tail = new TailContext(this);
head = new HeadContext(this);
head.next = tail;
tail.prev = head;
}`

这里实例化了 tail 和 head 这两个 handler。tail 实现了 ChannelInboundHandler 接口，

而 head 实现了 ChannelOutboundHandler 和 ChannelInboundHandler 两个接口，

而且最后两行代码将 tail 和 head 链接起来:复制代码

注意，在不一样的版本中，源码也略有差别，head 不必定是 in + out，你们知道这点就行了。

还有，从上面的 head 和 tail 咱们也能够看到，其实 pipeline 中的每一个元素是 ChannelHandlerContext 的实例，而不是 ChannelHandler 的实例，context 包装了一下 handler，可是，后面咱们都会用 handler 来描述一个 pipeline 上的节点，而不是使用 context，但愿读者知道这一点。

这里只是构造了 pipeline，而且添加了两个固定的 handler 到其中（head + tail），还不涉及到自定义的 handler 代码执行。咱们回过头来看下面这段代码：

咱们说过 childHandler 中指定的 handler 不是给 NioServerSocketChannel 使用的，是给 NioSocketChannel 使用的，因此这里咱们不看它。

这里调用 handler(…) 方法指定了一个 LoggingHandler 的实例，而后咱们再进去下面的 bind(…) 方法中看看这个 LoggingHandler 实例是怎么进入到咱们以前构造的 pipeline 内的。

顺着 bind() 一直往前走，bind() -> doBind() -> initAndRegister()：

`final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 1. 构造 channel 实例，同时会构造 pipeline 实例，
// 如今 pipeline 中有 head 和 tail 两个 handler 了
channel = channelFactory.newChannel();
// 2. 看这里
init(channel);
} catch (Throwable t) {
......
}`

上面的两行代码，第一行实现了构造 channel 和 channel 内部的 pipeline，咱们来看第二行 init 代码：

`// ServerBootstrap：
@Override
void init(Channel channel) throws Exception {
......
// 拿到刚刚建立的 channel 内部的 pipeline 实例
ChannelPipeline p = channel.pipeline();
...
// 开始往 pipeline 中添加一个 handler，这个 handler 是 ChannelInitializer 的实例
p.addLast(new ChannelInitializer<Channel>() {
// 咱们之后会看到，下面这个 initChannel 方法什么时候会被调用
@Override
public void initChannel(final Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
// 这个方法返回咱们最开始指定的 LoggingHandler 实例
ChannelHandler handler = config.handler();
if (handler != null) {
// 添加 LoggingHandler
pipeline.addLast(handler);
}
// 先不用管这里的 eventLoop
ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
// 添加一个 handler 到 pipeline 中：ServerBootstrapAcceptor
// 从名字能够看到，这个 handler 的目的是用于接收客户端请求
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}
`
这里涉及到 pipeline 中的辅助类 ChannelInitializer，咱们看到，它自己是一个 handler（Inbound 类型），可是它的做用和普通 handler 有点不同，它纯碎是用来辅助将其余的 handler 加入到 pipeline 中的。

你们能够稍微看一下 ChannelInitializer 的 initChannel 方法，有个简单的认识就好，此时的 pipeline 应该是这样的：

ChannelInitializer 的 initChannel(channel) 方法被调用的时候，会往 pipeline 中添加咱们最开始指定的 LoggingHandler 和添加一个 ServerBootstrapAcceptor。可是咱们如今还不知道这个 initChannel 方法什么时候会被调用。

上面咱们说的是做为服务端的 NioServerSocketChannel 的 pipeline，NioSocketChannel 也是差很少的，咱们能够看一下 Bootstrap 类的 init(channel) 方法：

`void init(Channel channel) throws Exception {
ChannelPipeline p = channel.pipeline();
p.addLast(config.handler());
...
}`

它和服务端 ServerBootstrap 要添加 ServerBootstrapAcceptor 不同，它只须要将 EchoClient 类中的 ChannelInitializer 实例加进来就能够了，它的 ChannelInitializer 中添加了两个 handler，LoggingHandler 和 EchoClientHandler：

很显然，咱们须要的是像 LoggingHandler 和 EchoClientHandler 这样的 handler，可是，它们如今还不在 pipeline 中，那么它们何时会真正进入到 pipeline 中呢？之后咱们再揭晓。

还有，为何 Server 端咱们指定的是一个 handler 实例，而 Client 指定的是一个 ChannelInitializer 实例？其实它们是能够随意搭配使用的，你甚至能够在 ChannelInitializer 实例中添加 ChannelInitializer 的实例。

很是抱歉，这里又要断了，下面要先介绍线程池了，你们要记住 pipeline 如今的样子，head + channelInitializer + tail。

本节没有介绍 handler 的向后传播，就是一个 handler 处理完了之后，怎么传递给下一个 handler 来处理？好比咱们熟悉的 JavaEE 中的 Filter 是采用在一个 Filter 实例中调用 chain.doFilter(request, response) 来传递给下一个 Filter 这种方式的。

咱们用下面这张图结束本节。下图展现了传播的方法，但我实际上是更想让你们看一下，哪些事件是 Inbound 类型的，哪些是 Outbound 类型的：

Outbound 类型的几个事件你们应该比较好认，注意 bind 也是 Outbound 类型的。

365天干货不断微信搜索「猿灯塔」第一时间阅读，回复【资料】【面试】【简历】有我准备的一线大厂面试资料和简历模板