認真的 Netty 原始碼分析（二）
新的主題，右下角可以設定，歡迎大家提出建議。
認真的 Netty 原始碼解析（二）
更新時間：2018-11-03
Channel 的 register 操作
經過前面的鋪墊，我們已經具備一定的基礎了，我們開始來把前面學到的內容揉在一起。這節，我們會介紹 register 操作，這一步其實是非常關鍵的，對於我們原始碼分析非常重要。
register
我們從 EchoClient 中的 connect() 方法出發，或者 EchoServer 的 bind(port) 方法出發，都會走到 initAndRegister() 這個方法：
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
// 1
channel = channelFactory.newChannel();
// 2 對於 Bootstrap 和 ServerBootstrap，這裡面有些不一樣
init(channel);
} catch (Throwable t) {
...
}
// 3 我們這裡要說的是這行
ChannelFuture regFuture = config().group().register(channel);
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
return regFuture;
}
initAndRegister() 這個方法我們已經接觸過兩次了，前面介紹了 :one: Channel 的例項化，例項化過程中，會執行 Channel 內部 Unsafe 和 Pipeline 的例項化，以及在上面 :two: init(channel) 方法中，會往 pipeline 中新增 handler（pipeline 此時是 head+channelnitializer+tail）。
我們這節終於要揭祕 ChannelInitializer 中的 initChannel 方法了~~~
現在，我們繼續往下走，看看 :three: register 這一步：
ChannelFuture regFuture = config().group().register(channel);
我們說了，register 這一步是非常關鍵的，它發生在 channel 例項化以後，大家回憶一下當前 channel 中的一些情況：
例項化了 JDK 底層的 Channel，設定了非阻塞，例項化了 Unsafe，例項化了 Pipeline，同時往 pipeline 中添加了 head、tail 以及一個 ChannelInitializer 例項。
上面的 group() 方法會返回前面例項化的 NioEventLoopGroup 的例項，然後呼叫其 register(channel) 方法：
// MultithreadEventLoopGroup
@Override
public ChannelFuture register(Channel channel) {
return next().register(channel);
}
next() 方法很簡單，就是選擇執行緒池中的一個執行緒（還記得 chooserFactory 嗎），也就是選擇一個 NioEventLoop 例項，這個時候我們就進入到 NioEventLoop 了。
NioEventLoop 的 register(channel) 方法實現在它的父類 SingleThreadEventLoop 中：
@Override
public ChannelFuture register(Channel channel) {
return register(new DefaultChannelPromise(channel, this));
}
這裡例項化了一個 Promise，將當前 channel 帶進去：
@Override
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
// promise 關聯了 channel，channel 持有 Unsafe 例項，register 操作就封裝在 Unsafe 中
promise.channel().unsafe().register(this, promise);
return promise;
}
拿到 channel 中關聯的 Unsafe 例項，然後呼叫它的 register 方法：
我們說過，Unsafe 專門用來封裝底層實現，當然這裡也沒那麼“底層”
// AbstractChannel# AbstractUnsafe
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
...
// 將這個 eventLoop 例項設定給這個 channel，從此這個 channel 就是有 eventLoop 的了
// 我覺得這一步其實挺關鍵的，因為後續該 channel 中的所有非同步操作，都要提交給這個 eventLoop 來執行
AbstractChannel.this.eventLoop = eventLoop;
// 如果發起 register 動作的執行緒就是 eventLoop 例項中的執行緒，那麼直接呼叫 register0(promise)
// 對於我們來說，它不會進入到這個分支，之所以有這個分支，是因為我們是可以 unregister，然後再 register 的
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
// 否則，提交任務給 eventLoop，eventLoop 中的執行緒會負責呼叫 register0(promise)
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
...
}
}
}
到這裡，我們要明白，NioEventLoop 中是還沒有例項化 Thread 例項的。
這幾步涉及到了好幾個類：NioEventLoop、Promise、Channel、Unsafe 等，大家要仔細理清楚它們的關係。
對於我們前面過來的 register 操作，其實提交到 eventLoop 以後，就直接返回 promise 例項了，剩下的register0 是非同步操作。
我們這邊先不繼續往裡分析 register0(promise) 方法，先把前面欠下的 NioEventLoop 中的執行緒介紹清楚，然後再回來介紹這個 register0 方法。
NioEventLoop 工作流程
前面，我們在分析執行緒池的例項化的時候說過，NioEventLoop 中並沒有啟動 Java 執行緒。這裡我們來仔細分析下在 register 過程中呼叫的 eventLoop.execute(runnable) 這個方法，這個程式碼在父類 SingleThreadEventExecutor 中：
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
// 判斷新增任務的執行緒是否就是當前 EventLoop 中的執行緒
boolean inEventLoop = inEventLoop();
// 新增任務到之前介紹的 taskQueue 中，
//如果 taskQueue 滿了(預設大小 16)，根據我們之前說的，預設的策略是丟擲異常
addTask(task);
if (!inEventLoop) {
// 如果不是 NioEventLoop 內部執行緒提交的 task，那麼判斷下執行緒是否已經啟動，沒有的話，就啟動執行緒
startThread();
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
原來啟動 NioEventLoop 中的執行緒的方法在這裡
下面是 startThread 的原始碼，判斷執行緒是否已經啟動來決定是否要進行啟動操作：
private void startThread() {
if (state == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
try {
doStartThread();
} catch (Throwable cause) {
STATE_UPDATER.set(this, ST_NOT_STARTED);
PlatformDependent.throwException(cause);
}
}
}
}
我們按照前面的思路，根據執行緒沒有啟動的情況，來看看 doStartThread() 方法：
private void doStartThread() {
assert thread == null;
// 這裡的 executor 大家是不是有點熟悉的感覺，它就是一開始我們例項化 NioEventLoop 的時候傳進來的 ThreadPerTaskExecutor 的例項。它是每次來一個任務，建立一個執行緒的那種 executor。
// 一旦我們呼叫它的 execute 方法，它就會建立一個新的執行緒，所以這裡終於會建立 Thread 例項
executor.execute(new Runnable() {
@Override
public void run() {
// 看這裡，將 “executor” 中建立的這個執行緒設定為 NioEventLoop 的執行緒！！！
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {
// 執行 SingleThreadEventExecutor 的 run() 方法，它在 NioEventLoop 中實現了
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
// ... 我們直接忽略掉這裡的程式碼
}
}
});
}
上面執行緒啟動以後，會執行 NioEventLoop 中的 run() 方法，這是一個非常重要的方法，這個方法肯定是沒那麼容易結束的，必然是像 JDK 執行緒池的 Worker 那樣，不斷地迴圈獲取新的任務的。它需要不斷地做 select 操作和輪詢 taskQueue 這個佇列。
我們先來簡單地看一下它的原始碼，這裡先不做深入地介紹：
@Override
protected void run() {
// 程式碼巢狀在 for 迴圈中
for (;;) {
try {
// selectStrategy 終於要派上用場了
// 它有兩個值，一個是 CONTINUE 一個是 SELECT
// 針對這塊程式碼，我們分析一下，如果 taskQueue 不為空，也就是 hasTasks() 返回 true，
//那麼執行一次 selectNow()，因為該方法不會阻塞
// 如果 hasTasks() 返回 false，那麼執行 SelectStrategy.SELECT 分支，進行 select(...)，這塊是帶阻塞的
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
// 如果 !hasTasks()，那麼進到這個 select 分支，這裡 select 帶阻塞的
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
cancelledKeys = 0;
needsToSelectAgain = false;
// 預設地，ioRatio 的值是 50
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
// 如果 ioRatio 設定為 100，那麼先執行 IO 操作，然後在 finally 塊中執行 taskQueue 中的任務
try {
// 1. 執行 IO 操作。因為前面 select 以後，可能有些 channel 是需要處理的。
processSelectedKeys();
} finally {
// 2. 執行非 IO 任務，也就是 taskQueue 中的任務
runAllTasks();
}
} else {
// 如果 ioRatio 不是 100，那麼根據 IO 操作耗時，限制非 IO 操作耗時
final long ioStartTime = System.nanoTime();
try {
// 執行 IO 操作
processSelectedKeys();
} finally {
// 根據 IO 操作消耗的時間，計算執行非 IO 操作（runAllTasks）可以用多少時間.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
上面這段程式碼是 NioEventLoop 的核心，這裡介紹兩點：
- 首先，會根據 hasTasks() 的結果來決定是執行 selectNow() 還是 select(oldWakenUp)，這個應該好理解。如果有任務正在等待，那麼應該使用無阻塞的 selectNow()，如果沒有任務在等待，那麼就可以使用帶阻塞的 select 操作。
- ioRatio 控制 IO 操作所佔的時間比重：
- 如果設定為 100%，那麼先執行 IO 操作，然後再執行任務佇列中的任務。
- 如果不是 100%，那麼先執行 IO 操作，然後執行 taskQueue 中的任務，但是需要控制執行任務的總時間，這個時間通過 ioRatio，以及這次 IO 操作耗時計算得出。
我們這裡先不要去關心 select(oldWakenUp)、processSelectedKeys() 方法和 runAllTasks(…) 方法的細節，只要先理解它們分別做什麼事情就可以了。
回過神來，我們前面在 register 的時候提交了 register 任務給 NioEventLoop，這是 NioEventLoop 接收到的第一個任務，所以這裡會例項化 Thread 並且啟動，然後進入到 NioEventLoop 中的 run 方法。
繼續 register
我們回到前面的 register0(promise) 方法，我們知道，這個 register 任務進入到了 NioEventLoop 的 taskQueue 中，然後會啟動 NioEventLoop 中的執行緒，該執行緒會輪詢這個 taskQueue，然後執行這個 register 任務。所以，此時執行該方法的是 eventLoop 中的執行緒：
// AbstractChannel
private void register0(ChannelPromise promise) {
try {
...
boolean firstRegistration = neverRegistered;
// *** 進行 JDK 底層的操作：Channel 註冊到 Selector 上 ***
doRegister();
neverRegistered = false;
registered = true;
// 到這裡，就算是 registered 了
// 這一步也很關鍵，因為這涉及到了 ChannelInitializer 的 init(channel)
// 我們之前說過，init 方法會將 ChannelInitializer 內部新增的 handlers 新增到 pipeline 中
pipeline.invokeHandlerAddedIfNeeded();
// 設定當前 promise 的狀態為 success
//因為當前 register 方法是在 eventLoop 中的執行緒中執行的，需要通知提交 register 操作的執行緒
safeSetSuccess(promise);
// 當前的 register 操作已經成功，該事件應該被 pipeline 上
//所有關心 register 事件的 handler 感知到，往 pipeline 中扔一個事件
pipeline.fireChannelRegistered();
// 這裡 active 指的是 channel 已經開啟
if (isActive()) {
// 如果該 channel 是第一次執行 register，那麼 fire ChannelActive 事件
if (firstRegistration) {
pipeline.fireChannelActive();
} else if (config().isAutoRead()) {
// 該 channel 之前已經 register 過了，
// 這裡讓該 channel 立馬去監聽通道中的 OP_READ 事件
beginRead();
}
}
} catch (Throwable t) {
...
}
}
我們先說掉上面的 doRegister() 方法，然後再說 pipeline。
@Override
protected void doRegister() throws Exception {
boolean selected = false;
for (;;) {
try {
// 附 JDK 中 Channel 的 register 方法：
// public final SelectionKey register(Selector sel, int ops, Object att) {...}
selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException e) {
...
}
}
}
我們可以看到，這裡做了 JDK 底層的 register 操作，將 SocketChannel(或 ServerSocketChannel) 註冊到 Selector 中，並且可以看到，這裡的監聽集合設定為了 0 ，也就是什麼都不監聽。
當然，也就意味著，後續一定有某個地方會需要修改這個 selectionKey 的監聽集合。
我們重點來說說 pipeline 操作，我們之前在介紹 NioSocketChannel 的 pipeline 的時候介紹到，我們的 pipeline 現在長這個樣子：
現在，我們將看到這裡會把 LoggingHandler 和 EchoClientHandler 新增到 pipeline。
我們繼續看程式碼，register 成功以後，執行了以下操作：
pipeline.invokeHandlerAddedIfNeeded();
大家可以跟蹤一下，這一步會執行到 pipeline 中 ChannelInitializer 例項的 handlerAdded 方法，在這裡會執行它的 init(context) 方法：
@Override
public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
if (ctx.channel().isRegistered()) {
initChannel(ctx);
}
}
然後我們看下 initChannel(ctx)，這裡終於來了我們之前介紹過的 init(channel) 方法：
private boolean initChannel(ChannelHandlerContext ctx) throws Exception {
if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { // Guard against re-entrance.
try {
// 1. 將把我們自定義的 handlers 新增到 pipeline 中
initChannel((C) ctx.channel());
} catch (Throwable cause) {
...
} finally {
// 2. 將 ChannelInitializer 例項從 pipeline 中刪除
remove(ctx);
}
return true;
}
return false;
}
我們前面也說過，ChannelInitializer 的 init(channel) 被執行以後，那麼其內部新增的 handlers 會進入到 pipeline 中，然後上面的 finally 塊中將 ChannelInitializer 的例項從 pipeline 中刪除，那麼此時 pipeline 就算建立起來了，如下圖：
其實這裡還有個問題，如果我們在 ChannelInitializer 中又新增 ChannelInitializer 呢？大家可以考慮下這個情況。
pipeline 建立了以後，然後我們繼續往下走，會執行到這一句：
pipeline.fireChannelRegistered();
我們只要摸清楚了 fireChannelRegistered() 方法，以後碰到其他像 fireChannelActive()、fireXxx() 等就知道怎麼回事了，它們都是類似的。我們來看看這句程式碼會發生什麼：
// DefaultChannelPipeline
@Override
public final ChannelPipeline fireChannelRegistered() {
// 注意這裡的傳參是 head
AbstractChannelHandlerContext.invokeChannelRegistered(head);
return this;
}
也就是說，我們往 pipeline 中扔了一個 channelRegistered 事件，這裡的 register 屬於 Inbound 事件，pipeline 接下來要做的就是執行 pipeline 中的 Inbound handlers 中的 channelRegistered() 方法。
從上面的程式碼，我們可以看出，往 pipeline 中扔出 channelRegistered 事件以後，第一個處理的 handler 是 head 。
接下來，我們還是跟著程式碼走，此時我們來到了 pipeline 的第一個節點 head 的處理中：
// AbstractChannelHandlerContext
// next 此時是 head
static void invokeChannelRegistered(final AbstractChannelHandlerContext next) {
EventExecutor executor = next.executor();
// 執行 head 的 invokeChannelRegistered()
if (executor.inEventLoop()) {
next.invokeChannelRegistered();
} else {
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelRegistered();
}
});
}
}
也就是說，這裡會先執行 head.invokeChannelRegistered() 方法，而且是放到 NioEventLoop 中的 taskQueue 中執行的：
// AbstractChannelHandlerContext
private void invokeChannelRegistered() {
if (invokeHandler()) {
try {
// handler() 方法此時會返回 head
((ChannelInboundHandler) handler()).channelRegistered(this);
} catch (Throwable t) {
notifyHandlerException(t);
}
} else {
fireChannelRegistered();
}
}
我們去看 head 的 channelRegistered 方法：
// HeadContext
@Override
public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
// 1. 這一步是 head 對於 channelRegistered 事件的處理。沒有我們要關心的
invokeHandlerAddedIfNeeded();
// 2. 向後傳播 Inbound 事件
ctx.fireChannelRegistered();
}
然後 head 會執行 fireChannelRegister() 方法：
// AbstractChannelHandlerContext
@Override
public ChannelHandlerContext fireChannelRegistered() {
// 這裡很關鍵
// findContextInbound() 方法會沿著 pipeline 找到下一個 Inbound 型別的 handler
invokeChannelRegistered(findContextInbound());
return this;
}
注意：pipeline.fireChannelRegistered() 是將 channelRegistered 事件拋到 pipeline 中，pipeline 中的 handlers 準備處理該事件。而 context.fireChannelRegistered() 是一個 handler 處理完了以後，向後傳播給下一個 handler。
它們兩個的方法名字是一樣的，但是來自於不同的類。
findContextInbound() 將找到下一個 Inbound 型別的 handler，然後又是重複上面的幾個方法。
我覺得上面這塊程式碼沒必要太糾結，總之就是從 head 中開始，依次往下尋找所有 Inbound handler，執行其 channelRegistered(ctx) 操作。
說了這麼多，我們的 register 操作算是真正完成了。
下面，我們回到 initAndRegister 這個方法：
final ChannelFuture initAndRegister() {
Channel channel = null;
try {
channel = channelFactory.newChannel();
init(channel);
} catch (Throwable t) {
...
}
// 我們上面說完了這行
ChannelFuture regFuture = config().group().register(channel);
// 如果在 register 的過程中，發生了錯誤
if (regFuture.cause() != null) {
if (channel.isRegistered()) {
channel.close();
} else {
channel.unsafe().closeForcibly();
}
}
// 原始碼中說得很清楚，如果到這裡，說明後續可以進行 connect() 或 bind() 了，因為兩種情況：
// 1. 如果 register 動作是在 eventLoop 中發起的，那麼到這裡的時候，register 一定已經完成
// 2. 如果 register 任務已經提交到 eventLoop 中，也就是進到了 eventLoop 中的 taskQueue 中，
//由於後續的 connect 或 bind 也會進入到同一個 eventLoop 的 queue 中，所以一定是會先 register 成功，才會執行 connect 或 bind
return regFuture;
}
我們要知道，不管是服務端的 NioServerSocketChannel 還是客戶端的 NioSocketChannel，在 bind 或 connect 時，都會先進入 initAndRegister 這個方法，所以我們上面說的那些，對於兩者都是通用的。
大家要記住，register 操作是非常重要的，要知道這一步大概做了哪些事情，register 操作以後，將進入到 bind 或 connect 操作中。
connect 過程和 bind 過程分析
上面我們介紹的 register 操作非常關鍵，它建立起來了很多的東西，它是 Netty 中 NioSocketChannel 和 NioServerSocketChannel 開始工作的起點。
這一節，我們來說說 register 之後的 connect 操作和 bind 操作。
connect 過程分析
對於客戶端 NioSocketChannel 來說，前面 register 完成以後，就要開始 connect 了，這一步將連線到服務端。
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
// 這裡完成了 register 操作
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
// 這裡我們不去糾結 register 操作是否 isDone()
if (regFuture.isDone()) {
if (!regFuture.isSuccess()) {
return regFuture;
}
// 看這裡
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else {
....
}
}
這裡大家自己一路點進去，我就不浪費篇幅了。最後，我們會來到 AbstractChannel 的 connect 方法：
@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return pipeline.connect(remoteAddress, promise);
}
我們看到，connect 操作是交給 pipeline 來執行的。進入 pipeline 中，我們會發現，connect 這種 Outbound 型別的操作，是從 pipeline 的 tail 開始的：
前面我們介紹的 register 操作是 Inbound 的，是從 head 開始的
@Override
public final ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
return tail.connect(remoteAddress, promise);
}
接下來就是 pipeline 的操作了，從 tail 開始，執行 pipeline 上的 Outbound 型別的 handlers 的 connect(...) 方法，那麼真正的底層的 connect 的操作發生在哪裡呢？還記得我們的 pipeline 的圖嗎？
從 tail 開始往前找 out 型別的 handlers，最後會到 head 中，因為 head 也是 Outbound 型別的，我們需要的 connect 操作就在 head 中，它會負責呼叫 unsafe 中提供的 connect 方法：
// HeadContext
public void connect(
ChannelHandlerContext ctx,
SocketAddress remoteAddress, SocketAddress localAddress,
ChannelPromise promise) throws Exception {
unsafe.connect(remoteAddress, localAddress, promise);
}
接下來，我們來看一看 connect 在 unsafe 類中所謂的底層操作：
// AbstractNioChannel.AbstractNioUnsafe
@Override
public final void connect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
......
boolean wasActive = isActive();
// 大家自己點進去看 doConnect 方法
// 這一步會做 JDK 底層的 SocketChannel connect，然後設定 interestOps 為 SelectionKey.OP_CONNECT
// 返回值代表是否已經連線成功
if (doConnect(remoteAddress, localAddress)) {
// 處理連線成功的情況
fulfillConnectPromise(promise, wasActive);
} else {
connectPromise = promise;
requestedRemoteAddress = remoteAddress;
// 下面這塊程式碼，在處理連線超時的情況，程式碼很簡單
// 這裡用到了 NioEventLoop 的定時任務的功能，這個我們之前一直都沒有介紹過，因為我覺得也不太重要
int connectTimeoutMillis = config().getConnectTimeoutMillis();
if (connectTimeoutMillis > 0) {
connectTimeoutFuture = eventLoop().schedule(new Runnable() {
@Override
public void run() {
ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
ConnectTimeoutException cause =
new ConnectTimeoutException("connection timed out: " + remoteAddress);
if (connectPromise != null && connectPromise.tryFailure(cause)) {
close(voidPromise());
}
}
}, connectTimeoutMillis, TimeUnit.MILLISECONDS);
}
promise.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (future.isCancelled()) {
if (connectTimeoutFuture != null) {
connectTimeoutFuture.cancel(false);
}
connectPromise = null;
close(voidPromise());
}
}
});
}
} catch (Throwable t) {
promise.tryFailure(annotateConnectException(t, remoteAddress));
closeIfClosed();
}
}
如果上面的 doConnect 方法返回 false，那麼後續是怎麼處理的呢？
在上一節介紹的 register 操作中，channel 已經 register 到了 selector 上，只不過將 interestOps 設定為了 0，也就是什麼都不監聽。
而在上面的 doConnect 方法中，我們看到它在呼叫底層的 connect 方法後，會設定 interestOps 為 SelectionKey.OP_CONNECT。
剩下的就是 NioEventLoop 的事情了，還記得 NioEventLoop 的 run() 方法嗎？也就是說這裡的 connect 成功以後，會在 run() 方法中被 processSelectedKeys() 方法處理掉。
bind 過程分析
說完 connect 過程，我們再來簡單看下 bind 過程：
private ChannelFuture doBind(final SocketAddress localAddress) {
// **前面說的 initAndRegister**
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
if (regFuture.cause() != null) {
return regFuture;
}
if (regFuture.isDone()) {
// register 動作已經完成，那麼執行 bind 操作
ChannelPromise promise = channel.newPromise();
doBind0(regFuture, channel, localAddress, promise);
return promise;
} else {
......
}
}
然後一直往裡看，會看到，bind 操作也是要由 pipeline 來完成的：
// AbstractChannel
@Override
public ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return pipeline.bind(localAddress, promise);
}
bind 操作和 connect 一樣，都是 Outbound 型別的，所以都是 tail 開始：
@Override
public final ChannelFuture bind(SocketAddress localAddress, ChannelPromise promise) {
return tail.bind(localAddress, promise);
}
最後的 bind 操作又到了 head 中，由 head 來呼叫 unsafe 提供的 bind 方法：
@Override
public void bind(
ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise)
throws Exception {
unsafe.bind(localAddress, promise);
}
感興趣的讀者自己去看一下 unsafe 中的 bind 方法，非常簡單，bind 操作也不是什麼非同步方法，我們就介紹到這裡了。
本節非常簡單，就是想和大家介紹下 Netty 中各種操作的套路。