Netty源码分析--认真系列(一)

相关文章链接

位运算详解

waken方法详解

ThreadPerTaskExecutor与线程创建详解

processSelectedKeys() vs runAllTasks()

NioServerSocketChannel-Unsafe初始化详解

NioEventLoop的run方法详解

NioEventLoopGroup深度解析

inEventLoop方法详解

executionMask详解

Netty源码分析–认真系列(一)

Netty 源码分析笔记 – 总结

学习 Netty 源码也有一段时间了，这篇文章算是对这段时间学习的一个总结。主要记录一下 Netty 的核心原理和工作流程，方便以后复习，也希望能帮到正在学习 Netty 的朋友们。

说明：本文基于 Netty 4.1.36 版本

为什么用 Netty？

Java 网络编程有三种 I/O 模型：BIO、NIO、AIO。BIO 是阻塞 IO，一个连接一个线程，并发能力差；AIO 是异步 IO，理论上性能最好，但实际用的人很少。实际开发中基本都用 NIO，因为它在性能和易用性之间取得了平衡。

不过原生 NIO 用起来真的很痛苦，代码写起来又臭又长。Netty 就是对 NIO 的封装，让我们能更方便地开发高性能的网络程序。除了简化 API，Netty 还做了很多性能优化，比如零拷贝、内存池等，也修复了 NIO 的一些 Bug（比如臭名昭著的 Epoll Bug）。

网络通信的本质

网络通信说白了就是客户端和服务端互相收发消息。Netty 提供了几个核心组件来实现这个过程：

Channel 就是一个连接，可以理解成一条电话线，数据通过它双向传输。

EventLoop 负责处理 Channel 上的 I/O 事件，比如读、写、连接等。它本质上就是一个线程，一个 EventLoop 可以管理多个 Channel，但一个 Channel 只会绑定到一个 EventLoop，这样就保证了线程安全。就像餐厅服务员，一个人可以服务多桌客人。

EventLoopGroup 就是线程池，管理多个 EventLoop。新连接来了，就轮询分配给某个 EventLoop 去处理。

ChannelHandler 是具体干活的，负责处理 I/O 事件或者拦截 I/O 操作。比如接收到数据后怎么处理，发送数据前要做什么转换，这些逻辑都写在 Handler 里。Handler 分两种：ChannelInboundHandler 处理入站事件（接收数据），ChannelOutboundHandler 处理出站事件（发送数据）。

ChannelPipeline 是处理链，数据进来或出去都要经过这条链上的多个 Handler。入站（接收数据）从 head 到 tail，出站（发送数据）从 tail 到 head。就像工厂流水线，每个 Handler 负责一道工序。

编解码器 负责在字节流和 Java 对象之间转换。网络传输只能用字节流，但业务代码用的是对象，所以需要编解码器来转换。解码器把字节流变成对象（接收时用），编码器把对象变成字节流（发送时用）。

Netty 核心组件关系图

┌─────────────────────────────────────────────────────────┐
│ Netty 应用程序 │
│ │
│ ┌────────────────────────────────────────────────┐ │
│ │ EventLoopGroup（线程池） │ │
│ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │
│ │ │EventLoop1│ │EventLoop2│ │EventLoop3│ │ │
│ │ │(线程1) │ │(线程2) │ │(线程3) │ │ │
│ │ └────┬─────┘ └────┬─────┘ └────┬─────┘ │ │
│ └───────┼─────────────┼─────────────┼───────────┘ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │Channel1 │ │Channel2 │ │Channel3 │ │
│ │(连接1) │ │(连接2) │ │(连接3) │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │
│ ↓ ↓ ↓ │
│ ┌─────────────────────────────────────┐ │
│ │ ChannelPipeline（处理链） │ │
│ │ ┌────────┐ ┌────────┐ ┌────────┐│ │
│ │ │Decoder │→ │Business│→ │Encoder ││ │
│ │ │(解码器)│ │Handler │ │(编码器)││ │
│ │ └────────┘ └────────┘ └────────┘│ │
│ └─────────────────────────────────────┘ │
│ │
│ ↕ 字节流 │
│ ┌─────────────┐ │
│ │ Socket │ │
│ │ (网络连接) │ │
│ └─────────────┘ │
└─────────────────────────────────────────────────────────┘

模板代码

上面是 Netty 的基本模板代码，每次使用都是这个套路。唯一需要我们开发的部分是 handler(…) 和 childHandler(…) 方法中指定的各个 handler，Netty 源码给我们提供了很多现成的 handler，需要的时候直接拿过来用就好了。

Netty 中的 Channel

我们先来看看 Netty 中的 Channel。在上面的模板代码中可以看到我们用的是 NioSocketChannel、NioServerSocketChannel，底层还是要用到 JDK 的 SocketChannel，我们来看看它们是怎么联系在一起的。

在 Bootstrap（客户端）和 ServerBootstrap（服务端）的启动过程中都会调用 channel(…) 方法，我们来看看 channel 源码：

public B channel(Class<? extends C> channelClass) {
if (channelClass == null) {
throw new NullPointerException("channelClass");
} else {
//在这里会继续调用channelFactory()方法
return (B)this.channelFactory(new ReflectiveChannelFactory(channelClass));
}
}

public B channelFactory(io.netty.channel.ChannelFactory<? extends C> channelFactory) {
return (B)this.channelFactory(channelFactory);
}

@Deprecated
public B channelFactory(ChannelFactory<? extends C> channelFactory) {
if (channelFactory == null) {
throw new NullPointerException("channelFactory");
} else if (this.channelFactory != null) {
throw new IllegalStateException("channelFactory set already");
} else {
this.channelFactory = channelFactory;
return (B)this.self();
}
}

我们可以看到，这个方法设置了 channelFactory 为 ReflectiveChannelFactory 的一个实例，接下来看看 ReflectiveChannelFactory 到底是什么：

public class ReflectiveChannelFactory<T extends Channel> implements ChannelFactory<T> {

private final Constructor<? extends T> constructor;

public ReflectiveChannelFactory(Class<? extends T> clazz) {
ObjectUtil.checkNotNull(clazz, "clazz");
try {
//看到这里，获取传入类的无参构造函数，赋值给constructor
this.constructor = clazz.getConstructor();
} catch (NoSuchMethodException e) {
throw new IllegalArgumentException("Class " + StringUtil.simpleClassName(clazz) +
" does not have a public non-arg constructor", e);
}
}

@Override
public T newChannel() {
try {
//在newChannel方法中调用传入类的newInstance方法
return constructor.newInstance();
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + constructor.getDeclaringClass(), t);
}
}

在构造方法中，设置 constructor 为传入类的无参构造函数。在上面的案例中也就是设置 constructor 为 NioSocketChannel.class 类的无参构造函数。这在后面调用 connect(…)、bind(…) 方法时会用到，用于 channel 的初始化。

接下来，我们简单追踪下充当客户端的 Bootstrap 中 NioSocketChannel 的创建过程，看看 NioSocketChannel 是怎么和 JDK 中的 SocketChannel 关联在一起的：

public ChannelFuture connect(String inetHost, int inetPort) {
return this.connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}

public ChannelFuture connect(InetAddress inetHost, int inetPort) {
return this.connect(new InetSocketAddress(inetHost, inetPort));
}

public ChannelFuture connect(SocketAddress remoteAddress) {
if (remoteAddress == null) {
throw new NullPointerException("remoteAddress");
} else {
// validate 只是校验一下各个参数是不是正确设置了
this.validate();
//主要关注doResolveAndConnect()
return this.doResolveAndConnect(remoteAddress, this.config.localAddress());
}
}

private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
//关注initAndRegister()
ChannelFuture regFuture = this.initAndRegister();
final Channel channel = regFuture.channel();
............
}

final ChannelFuture initAndRegister() {
Channel channel = null;
try {
//找到这里
channel = this.channelFactory.newChannel();
this.init(channel);
} catch (Throwable t) {
}
}
return regFuture;
}

看 channel = this.channelFactory.newChannel(); 这行代码，这里的 this.channelFactory 就是上面 channel.channelFactory() 方法设置的值。对应上面的案例，这里的值为 ReflectiveChannelFactory，也就是调用 ReflectiveChannelFactory.newChannel() 方法，回到 ReflectiveChannelFactory 类中。

public T newChannel() {
try {
return (T)(this.constructor.newInstance());
} catch (Throwable t) {
throw new ChannelException("Unable to create Channel from class " + this.constructor.getDeclaringClass(), t);
}
}

这里的 this.constructor 是在 ReflectiveChannelFactory 类的构造函数中设置的。对应上面案例，this.constructor == NioSocketChannel，也就是调用 NioSocketChannel 类的无参构造方法。

然后我们就可以去看 NioSocketChannel 的构造方法了：

private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();

public NioSocketChannel() {
//// SelectorProvider 实例用于创建 JDK 的 SocketChannel 实例
this(DEFAULT_SELECTOR_PROVIDER);
}

public NioSocketChannel(SelectorProvider provider) {
// // 看这里，newSocket(provider) 方法会创建 JDK 的 SocketChannel
this(newSocket(provider));
}

SelectorProvider 是一个抽象工厂类，在这里可以简单理解，SelectorProvider 就是用来创建 Selector、ServerSocketChannel、SocketChannel 的。

SelectorProvider.provider() 方法会根据操作系统返回对应的实现。例如 Windows 系统会返回 WindowsSelectorProvider，使用这里的 DEFAULT_SELECTOR_PROVIDER 就是 WindowsSelectorProvider 实例。

我们可以看到，在调用 newSocket(provider) 的时候，会创建 JDK NIO 的一个 SocketChannel（客户端连接通道）实例：

private static java.nio.channels.SocketChannel newSocket(SelectorProvider provider) {
try {
// 创建 SocketChannel 实例
return provider.openSocketChannel();
} catch (IOException e) {
throw new ChannelException("Failed to open a socket.", e);
}
}

NioServerSocketChannel 同理，也非常简单，从 ServerBootstrap#bind(…) 方法一路点进去就清楚了。

所以我们知道了，NioSocketChannel 在实例化过程中，会先实例化 JDK 底层的 SocketChannel，NioServerSocketChannel 也一样，会先实例化 ServerSocketChannel 实例：

上面我们知道了 Netty 中的 NioSocketChannel 是如何与 JDK 中的 SocketChannel 实例联系在一起的，但是 NioSocketChannel 构造函数中还有一个重要的地方没讲，我们接下来继续看 NioSocketChannel 构造函数。

public NioSocketChannel() {
this(DEFAULT_SELECTOR_PROVIDER);
}

public NioSocketChannel(SelectorProvider provider) {
this(newSocket(provider));
}

public NioSocketChannel(SocketChannel socket) {
this(null, socket);
}

public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
config = new NioSocketChannelConfig(this, socket.socket());
}

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
super(parent, ch, SelectionKey.OP_READ);
}

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
.................
}

protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}

一直往下追踪代码会到 AbstractChannel 类的构造函数中，我们来看下面这行代码，当前实际对象是 NioSocketChannel，所以会调用 NioSocketChannel 类实现的 newUnsafe() 方法：

unsafe = newUnsafe();

protected AbstractNioUnsafe newUnsafe() {
return new NioSocketChannelUnsafe();
}

所以 NioSocketChannel 对象中参数 unsafe 的值 = NioSocketChannelUnsafe 实例。

注意我这里说的版本是 4.1.36，在 4.1.x 早期版本中 NioSocketChannel 并没有 newUnsafe() 方法的实现，网上也有很多文章在这里说 unsafe 的值为 NioByteUnsafe 实例，这是因为版本不同。这里更改的原因是 SO_LINGER 导致 EventLoop 自旋（CPU 100%）问题。

这点很重要，在后续的 register 阶段会用到这个参数。

NioSocketChannel

//待完善

ChannelPipeline

上面我们了解了 channel 的初始化，通信工具有了，我们现在来讲讲 ChannelPipeline。

什么是 Pipeline？

我们在上面说 ChannelPipeline 是处理链，这里来简单说一下 pipeline 的概念。

Pipeline 是 Netty 中的责任链模式实现，它维护了一个由多个 ChannelHandler 组成的双向链表。每个 Channel 都有自己的 Pipeline，用于处理入站（Inbound）和出站（Outbound）事件。

Pipeline 的基本结构如下：

head <-> handler1 <-> handler2 <-> … <-> tail

其中 head 和 tail 是 Netty 自动创建的两个特殊节点，我们添加的 Handler 都会插入到它们之间。

ChannelPipeline 初始化

上面我们聊了 channel 的初始化，在 connect() 方法中会调用 NioSocketChannel 类的无参构造函数。

我们来看看 NioSocketChannel 的构造方法：

private static final SelectorProvider DEFAULT_SELECTOR_PROVIDER = SelectorProvider.provider();

public NioSocketChannel() {
//// SelectorProvider 实例用于创建 JDK 的 SocketChannel 实例
this(DEFAULT_SELECTOR_PROVIDER);
}

public NioSocketChannel(SelectorProvider provider) {
this(newSocket(provider));
}

public NioSocketChannel(SocketChannel socket) {
this(null, socket);
}
public NioSocketChannel(Channel parent, SocketChannel socket) {
super(parent, socket);
config = new NioSocketChannelConfig(this, socket.socket());
}

protected AbstractNioByteChannel(Channel parent, SelectableChannel ch) {
super(parent, ch, SelectionKey.OP_READ);
}

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
super(parent);
this.ch = ch;
this.readInterestOp = readInterestOp;
...........................
}

protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
pipeline = newChannelPipeline();
}

protected AbstractChannel(Channel parent) {
this.parent = parent;
id = newId();
unsafe = newUnsafe();
//看到这里，调用newChannelPipeline()方法初始化ChannelPipeline
pipeline = newChannelPipeline();
}

protected DefaultChannelPipeline newChannelPipeline() {
return new DefaultChannelPipeline(this);
}

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;
}

一直追踪下去会看见调用 newChannelPipeline()，返回 DefaultChannelPipeline 实例赋值给 pipeline，我们来看看 DefaultChannelPipeline 的构造函数。

在 DefaultChannelPipeline 类的构造函数中可以看出来，这里初始化了一个双向链表，创建头节点和尾节点，把头尾连起来，形成双向链表。

此时 Pipeline 的状态为：

head <-> tail

到这里就完成了 ChannelPipeline 的初始化，我们接下来看看如何将 ChannelHandler 加入 ChannelPipeline。

ChannelHandler 加入 ChannelPipeline

在这里我们来看看在 Netty 中是如何将 ChannelHandler 加入到 ChannelPipeline 中的。

我们视角回到模板代码中：

Pipeline 的初始化主要分为两个阶段：

为什么要分两个阶段呢？因为在 Channel 还没有注册到 EventLoop 之前，很多操作是不安全的。Netty 采用了一种"延迟初始化"的策略，先把需要执行的任务记录下来，等到合适的时机再统一执行。

第一阶段：准备阶段（Init）

我们以客户端 client 举例，首先看到 handler 方法：

public B handler(ChannelHandler handler) {
if (handler == null) {
throw new NullPointerException("handler");
}
this.handler = handler;
return self();
}

我们在 handler 方法中可以看见只是将我们的 handler（MyChannelInitializer）赋值给 handler 字段。

我们接着看 connect 方法，跟上面一样，一直追踪下去，追踪到 init 方法：

public ChannelFuture connect(String inetHost, int inetPort) {
return connect(InetSocketAddress.createUnresolved(inetHost, inetPort));
}

public ChannelFuture connect(SocketAddress remoteAddress) {
if (remoteAddress == null) {
throw new NullPointerException("remoteAddress");
}
// validate 只是校验一下各个参数是不是正确设置了
validate();
//主要关注doResolveAndConnect()
return doResolveAndConnect(remoteAddress, config.localAddress());
}

private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
//关注initAndRegister()
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
............
}

final ChannelFuture initAndRegister() {
Channel channel = null;
try {
channel = channelFactory.newChannel();
//看这里
init(channel);
} catch (Throwable t) {
}
return regFuture;
}

void init(Channel channel) throws Exception {
..............
//我们上面说过，在初始化阶段会调用NioSocketChannel类的无参构造函数，在NioSocketChannel类的无参构造函数中会初始化一个双向链表，这里获取的就是上面初始化赋值的pipeline
ChannelPipeline p = channel.pipeline();
................
//
p.addLast(new ChannelInitializer<Channel>() {
@Override
public void initChannel(final Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();
ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}

ch.eventLoop().execute(new Runnable() {
@Override
public void run() {
pipeline.addLast(new ServerBootstrapAcceptor(
ch, currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs));
}
});
}
});
}

我们来看到 init 方法，首先看到：

ChannelPipeline p = channel.pipeline();

我们点进去看 pipeline 方法：

public ChannelPipeline pipeline() {
return pipeline;
}

这里直接返回一个 pipeline，这个 pipeline 就是我们上面提到的 pipeline。在方法中会调用 NioSocketChannel 类的无参构造方法，会初始化一个双向链表，此时 pipeline 的状态为：

head <-> tail

接着我们来看 addLast 方法，这里会传入 ChannelInitializer 实例，这在下面 register 时会用到：

public final ChannelPipeline addLast(ChannelHandler... handlers) {
return addLast(null, handlers);
}

public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
if (handlers == null) {
throw new NullPointerException("handlers");
}
// 遍历每个 Handler，逐个添加
for (ChannelHandler h: handlers) {
if (h == null) {
break;
}
addLast(executor, null, h);
}

return this;
}

@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;

synchronized (this) {
// 检查 Handler 是否可以被多次添加
checkMultiplicity(handler);

// 步骤1：将 Handler 包装成 Context
newCtx = newContext(group, filterName(name, handler), handler);

// 步骤2：将 Context 添加到链表中
addLast0(newCtx);

// 步骤3：关键判断 – 如果还没有注册，创建待处理任务
if (!registered) {
newCtx.setAddPending();
// 创建 PendingHandlerAddedTask 并加入队列
callHandlerCallbackLater(newCtx, true);
return this;
}

// 如果已经注册，直接调用 handlerAdded
EventExecutor executor = newCtx.executor();
if (!executor.inEventLoop()) {
// … 省略
}
callHandlerAdded0(newCtx);
}
return this;
}

我们来看到 addLast 方法，这个方法是 Pipeline 初始化的核心，让我们逐步分析：

newCtx = newContext(group, filterName(name, handler), handler);

我们来看到 newContext，这个方法的作用是将 ChannelHandler 包装成 AbstractChannelHandlerContext。这个方法我们就不看源代码了，为什么要包装呢？因为 Pipeline 实际上是一个 Context 的双向链表，而不是 Handler 的链表。每个 Context 都持有一个 Handler 的引用，同时还包含了前后节点的引用、执行器等信息。简单来说，Handler 是业务逻辑处理器，而 Context 是 Handler 的容器，包含了 Handler 以及它在链表中的位置信息。

addLast0(newCtx);

我们看到 addLast0 的源代码：

private void addLast0(AbstractChannelHandlerContext newCtx) {
AbstractChannelHandlerContext prev = tail.prev;
newCtx.prev = prev;
newCtx.next = tail;
prev.next = newCtx;
tail.prev = newCtx;
}

作用很简单，就是将新创建的 Context 添加到 Pipeline 的双向链表末尾。

此时 pipeline 的状态：

head <-> ... <-> newCtx <-> tail

继续往下看到 callHandlerCallbackLater(newCtx, true)：

if (!registered) {
callHandlerCallbackLater(newCtx, true);
}

这里判断 Channel 是否已经注册到 EventLoop。为什么要这样做呢？因为在 Channel 还没有注册到 EventLoop 之前，很多操作是不安全的，例如：无法确定在哪个线程执行、EventLoop 还没有准备好。所以在这里先把需要执行的操作记录下来（创建任务），等到 Channel 注册完成后，再统一执行这些任务。

callHandlerCallbackLater() 方法的作用是创建待处理任务，我们来看到 callHandlerCallbackLater 方法源码：

private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) {
assert !registered;

// 根据 added 参数创建不同的任务
// added = true：创建 PendingHandlerAddedTask
// added = false：创建 PendingHandlerRemovedTask
PendingHandlerCallback task = added ?
new PendingHandlerAddedTask(ctx) :
new PendingHandlerRemovedTask(ctx);

// 将任务加入到待处理队列
PendingHandlerCallback pending = pendingHandlerCallbackHead;
if (pending == null) {
// 队列为空，设置为头节点
pendingHandlerCallbackHead = task;
} else {
// 队列不为空，添加到队列尾部
while (pending.next != null) {
pending = pending.next;
}
pending.next = task;
}
}

我们传入的 added 是 true，所以会创建 PendingHandlerAddedTask。

PendingHandlerAddedTask 就是一个简单的单向链表，是一个待处理的任务队列。

将创建的 PendingHandlerAddedTask 加入到待处理任务队列中。

到这里，准备阶段就完成了，待处理任务会在注册完成之后执行。

待处理任务队列：

pendingHandlerCallbackHead -> PendingHandlerAddedTask(ChannelInitializer)

接下来，我们进入第二阶段：执行阶段。

第二阶段：执行阶段（Register）

回到 initAndRegister() 方法，在调用完 init() 后，会继续执行注册操作：

final ChannelFuture initAndRegister() {
Channel channel = null;
try {
channel = channelFactory.newChannel();
init(channel); // 准备阶段完成
} catch (Throwable t) {
// …
}

// 执行阶段开始：注册 Channel
ChannelFuture regFuture = config().group().register(channel);
return regFuture;
}

config().group() 方法会返回前面实例化的 NioEventLoopGroup 的实例，然后调用其 register(channel) 方法。但是 NioEventLoopGroup 类没有 register(channel) 方法，所以会调用父类 MultithreadEventLoopGroup 的 register(channel) 方法：

public ChannelFuture register(Channel channel) {
return next().register(channel);
}

public EventLoop next() {
return (EventLoop) super.next();
}

public EventExecutor next() {
return chooser.next();
}

我们看到 next 方法，next 方法很简单，就是选择线程池中的一个线程，也就是选择一个 NioEventLoop 实例，这个时候我们就进入到 NioEventLoop 了。

这里的 chooser 就是 chooserFactory，chooserFactory 是一个线程选择器，后面我们会提到。详细介绍看 NioEventLoopGroup 深度解析.md 文档，在文档内部详细介绍了 NioEventLoopGroup 的初始化，在 NioEventLoopGroup 初始化过程中会涉及 chooserFactory 线程选择器的讲解。

我们来看到 NioEventLoop 的 register(channel) 方法，NioEventLoop 没有 register(channel) 方法，所以直接调用父类 SingleThreadEventLoop 的 register(channel) 方法：

// SingleThreadEventLoop.register()
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}

public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
}

promise 关联了 channel，channel 持有 Unsafe 实例，register 操作就封装在 Unsafe 中：

// AbstractChannel#AbstractUnsafe
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
// … 省略一些检查和设置

// 判断是否在 EventLoop 线程中
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
// 如果不在，提交任务到 EventLoop
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
}
}

接着看 register0() 方法。

doRegister() 方法会执行底层的注册操作，doRegister() 方法之后就会完成注册，所以后续的状态都是已注册状态。

private void register0(ChannelPromise promise) {
try {
// 确保 Channel 还是打开状态
if (!promise.setUncancellable() || !ensureOpen(promise)) {
return;
}
// 执行底层的注册操作（将 Channel 注册到 Selector）
doRegister();

neverRegistered = false;
registered = true;
// 关键：调用 Pipeline 的 invokeHandlerAddedIfNeeded()
pipeline.invokeHandlerAddedIfNeeded();
// … 其他逻辑
} catch (Throwable t) {
// … 异常处理
}
}

来看到 invokeHandlerAddedIfNeeded() 方法：

final void invokeHandlerAddedIfNeeded() {
assert channel.eventLoop().inEventLoop();

if (firstRegistration) {
firstRegistration = false;
// 调用所有待处理的 handlerAdded 回调
callHandlerAddedForAllHandlers();
}
}

private void callHandlerAddedForAllHandlers() {
final PendingHandlerCallback pendingHandlerCallbackHead;

synchronized (this) {
assert !registered;

// 标记为已注册
registered = true;

// 获取待处理任务队列的头节点
pendingHandlerCallbackHead = this.pendingHandlerCallbackHead;

// 清空队列（帮助 GC）
this.pendingHandlerCallbackHead = null;
}

// 遍历任务队列，执行每个任务
PendingHandlerCallback task = pendingHandlerCallbackHead;
while (task != null) {
task.execute(); // 执行任务
task = task.next; // 移动到下一个任务
}
}

@Override
void execute() {
EventExecutor executor = ctx.executor();
// 判断是否在 EventLoop 线程中
if (executor.inEventLoop()) {
// 在 EventLoop 线程中，直接调用
callHandlerAdded0(ctx);
} else {
// 不在 EventLoop 线程中，提交任务
try {
executor.execute(this);
} catch (RejectedExecutionException e) {
................
}
}
}

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
// 调用 Context 的 callHandlerAdded 方法
ctx.callHandlerAdded();
} catch (Throwable t) {
.................
}
}

final void callHandlerAdded() throws Exception {
// 设置状态为 ADD_COMPLETE
// 这个方法会检查状态，确保 handlerAdded 只被调用一次
if (setAddComplete()) {
// 调用 Handler 的 handlerAdded 方法
handler().handlerAdded(this);
}
}

一直追踪代码到：

handler().handlerAdded(this)

在这里会调用 handler 的 handlerAdded 方法，注意这里的 handler 是 ServerBootstrap.init() 方法中创建的匿名 ChannelInitializer：

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
// 检查 Channel 是否已经注册
if (ctx.channel().isRegistered()) {
// 调用 initChannel(ctx)
if (initChannel(ctx)) {
// 初始化成功后，移除自己
removeState(ctx);
}
}
}

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.add(ctx)) { // Guard against re-entrance.
try {
//注意看这个initChannel(Channel)
initChannel((C) ctx.channel());
} catch (Throwable cause) {
..............
} finally {
...............
}
return true;
}
return false;
}

追踪代码到 initChannel(ChannelHandlerContext)，在方法内部我们可以看到会调用 initChannel(Channel)。

我们回到 ServerBootstrap.init() 方法中，发现在这里重写了 initChannel(Channel)，所以这里调用的是 ServerBootstrap.init() 方法中创建的匿名类 ChannelInitializer 重写的 initChannel(Channel)：

// 这是在 Bootstrap.init() 中定义的匿名内部类
new ChannelInitializer<Channel>() {
@Override
public void initChannel(final Channel ch) throws Exception {
final ChannelPipeline pipeline = ch.pipeline();

// 获取用户设置的 handler（MyChannelInitializer）
ChannelHandler handler = config.handler();

if (handler != null) {
// 将 MyChannelInitializer 添加到 Pipeline
pipeline.addLast(handler);
}
................
}
}

我们来看到重写之后的 initChannel 方法：

final ChannelPipeline pipeline = ch.pipeline();

首先是获取配置的 pipeline。

ChannelHandler handler = config.handler();
if (handler != null) {
pipeline.addLast(handler);
}

然后是获取在客户端入口配置的 handler，也就是 MyChannelInitializer，将 MyChannelInitializer 加入到 pipeline 中。

重点看 addLast() 方法做了什么，又回到 addLast 方法中了：

@Override
public final ChannelPipeline addLast(ChannelHandler... handlers) {
return addLast(null, handlers);
}

public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
if (handlers == null) {
throw new NullPointerException("handlers");
}

for (ChannelHandler h: handlers) {
if (h == null) {
break;
}
// 遍历每个 Handler，逐个添加
addLast(executor, null, h);
}

return this;
}

@Override
public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
final AbstractChannelHandlerContext newCtx;

synchronized (this) {
// 检查 Handler 是否可以被多次添加
checkMultiplicity(handler);

// 将 MyChannelInitializer 包装成 Context（我们称它为 ctx-B）
newCtx = newContext(group, filterName(name, handler), handler);

// 将 ctx-B 添加到 Pipeline 链表中
addLast0(newCtx);

// 关键判断：此时 Channel 已经注册（registered = true）
if (!registered) {
// 如果还没注册，创建待处理任务
newCtx.setAddPending();
callHandlerCallbackLater(newCtx, true);
return this;
}

// 因为已经注册，所以直接调用 handlerAdded
EventExecutor executor = newCtx.executor();
if (!executor.inEventLoop()) {
// 如果不在 EventLoop 线程中，提交任务
callHandlerAddedInEventLoop(newCtx, executor);
return this;
}
}

// 在 EventLoop 线程中，直接调用
callHandlerAdded0(newCtx);
return this;
}

上面已经说过 addLast 方法了，在这里就不过多赘述了。

此时已经完成注册，会直接调用 callHandlerAdded0(newCtx) 方法：

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
try {
// 调用 callHandlerAdded
ctx.callHandlerAdded();
} catch (Throwable t) {
// 异常处理：如果出现异常，移除这个 Handler
}
}

final void callHandlerAdded() throws Exception {
// 设置状态为 ADD_COMPLETE，确保 handlerAdded 只被调用一次
if (setAddComplete()) {
// handler = MyChannelInitializer
handler().handlerAdded(this);
}
}

此时的 handler 为我们在客户端设置的 MyChannelInitializer：

public class MyChannelInitializer extends ChannelInitializer<SocketChannel> {

protected void initChannel(SocketChannel channel) throws Exception {
// 基于换行符号
channel.pipeline().addLast(new LineBasedFrameDecoder(1024));
// 解码转String，注意调整自己的编码格式GBK、UTF-8
channel.pipeline().addLast(new StringDecoder(Charset.forName("GBK")));
// 解码转String，注意调整自己的编码格式GBK、UTF-8
channel.pipeline().addLast(new StringEncoder(Charset.forName("GBK")));
// 在管道中添加我们自己的接收数据实现方法
channel.pipeline().addLast(new MyOutMsgHandler());//消息出站处理器，在Client发送消息时候会触发此处理器
channel.pipeline().addLast(new MyInMsgHandler()); //消息入站处理器
}

}

可以看到 MyChannelInitializer 继承 ChannelInitializer，并重写了 initChannel(SocketChannel) 方法。

MyChannelInitializer 未重写 handlerAdded(AbstractChannelHandlerContext) 方法，所以调用父类 ChannelInitializer 的 handlerAdded(AbstractChannelHandlerContext) 方法：

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
// 检查 Channel 是否已经注册
if (ctx.channel().isRegistered()) {
// 调用 initChannel(ctx)
if (initChannel(ctx)) {
// 初始化成功后，移除自己
removeState(ctx);
}
}
}

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.add(ctx)) { // Guard against re-entrance.
try {
//注意看这个initChannel(Channel)
initChannel((C) ctx.channel());
} catch (Throwable cause) {
..............
} finally {
...............
}
return true;
}
return false;
}

此时会调用 MyChannelInitializer 重写的 initChannel(SocketChannel) 方法。

看回到 MyChannelInitializer 重写的 initChannel(SocketChannel) 方法，会继续调用 addLast 方法将所有 handler 添加入 ChannelPipeline 中。

stractChannelHandlerContext) 方法，所以调用父类 ChannelInitializer 的 handlerAdded(AbstractChannelHandlerContext) 方法：

public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
// 检查 Channel 是否已经注册
if (ctx.channel().isRegistered()) {
// 调用 initChannel(ctx)
if (initChannel(ctx)) {
// 初始化成功后，移除自己
removeState(ctx);
}
}
}

private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.add(ctx)) { // Guard against re-entrance.
try {
//注意看这个initChannel(Channel)
initChannel((C) ctx.channel());
} catch (Throwable cause) {
..............
} finally {
...............
}
return true;
}
return false;
}

此时会调用 MyChannelInitializer 重写的 initChannel(SocketChannel) 方法。

看回到 MyChannelInitializer 重写的 initChannel(SocketChannel) 方法，会继续调用 addLast 方法将所有 handler 添加入 ChannelPipeline 中。

将 ChannelHandler 添加入 ChannelPipeline 的流程到此结束。