摘要：目錄源碼分析之番外篇的前生今世的前生今世之一簡介的前生今世之二小結的前生今世之三詳解的前生今世之四詳解源碼分析之零磨刀不誤砍柴工源碼分析環境搭建源碼分析之一揭開神秘的紅蓋頭源碼分析之一揭開神秘的紅蓋頭客戶端源碼分析之一揭開神秘的紅蓋頭服務器

目錄

Netty 源碼分析之 番外篇 Java NIO 的前生今世

Java NIO 的前生今世 之一 簡介

Java NIO 的前生今世 之二 NIO Channel 小結

Java NIO 的前生今世 之三 NIO Buffer 詳解

Java NIO 的前生今世 之四 NIO Selector 詳解

Netty 源碼分析之 零 磨刀不誤砍柴工 源碼分析環境搭建

Netty 源碼分析之 一 揭開 Bootstrap 神秘的紅蓋頭

Netty 源碼分析之 一 揭開 Bootstrap 神秘的紅蓋頭 (客戶端)

Netty 源碼分析之 一 揭開 Bootstrap 神秘的紅蓋頭 (服務器端)

這一章是 Netty 源碼分析系列 的第一章, 我打算在這一章中, 展示一下 Netty 的客戶端和服務端的初始化和啟動的流程, 給讀者一個對 Netty 源碼有一個大致的框架上的認識, 而不會深入每個功能模塊.
本章會從 Bootstrap/ServerBootstrap 類 入手, 分析 Netty 程序的初始化和啟動的流程.

Bootstrap 是 Netty 提供的一個便利的工廠類, 我們可以通過它來完成 Netty 的客戶端或服務器端的 Netty 初始化.
下面我以 Netty 源碼例子中的 Echo 服務器作為例子, 從客戶端和服務器端分別分析一下Netty 的程序是如何啟動的.

首先, 讓我們從客戶端方面的代碼開始
下面是源碼example/src/main/java/io/netty/example/echo/EchoClient.java 的客戶端部分的啟動代碼:

EventLoopGroup group = new NioEventLoopGroup();
try {
    Bootstrap b = new Bootstrap();
    b.group(group)
     .channel(NioSocketChannel.class)
     .option(ChannelOption.TCP_NODELAY, true)
     .handler(new ChannelInitializer() {
         @Override
         public void initChannel(SocketChannel ch) throws Exception {
             ChannelPipeline p = ch.pipeline();
             p.addLast(new EchoClientHandler());
         }
     });

    // Start the client.
    ChannelFuture f = b.connect(HOST, PORT).sync();

    // Wait until the connection is closed.
    f.channel().closeFuture().sync();
} finally {
    // Shut down the event loop to terminate all threads.
    group.shutdownGracefully();
}

從上面的客戶端代碼雖然簡單, 但是卻展示了 Netty 客戶端初始化時所需的所有內容:

EventLoopGroup: 不論是服務器端還是客戶端, 都必須指定 EventLoopGroup. 在這個例子中, 指定了 NioEventLoopGroup, 表示一個 NIO 的EventLoopGroup.

ChannelType: 指定 Channel 的類型. 因為是客戶端, 因此使用了 NioSocketChannel.

Handler: 設置數據的處理器.

下面我們深入代碼, 看一下客戶端通過 Bootstrap 啟動后, 都做了哪些工作.

在 Netty 中, Channel 是一個 Socket 的抽象, 它為用戶提供了關于 Socket 狀態(是否是連接還是斷開) 以及對 Socket 的讀寫等操作. 每當 Netty 建立了一個連接后, 都會有一個對應的 Channel 實例.
NioSocketChannel 的類層次結構如下:

這一小節我們著重分析一下 Channel 的初始化過程.

除了 TCP 協議以外, Netty 還支持很多其他的連接協議, 并且每種協議還有 NIO(異步 IO) 和 OIO(Old-IO, 即傳統的阻塞 IO) 版本的區別. 不同協議不同的阻塞類型的連接都有不同的 Channel 類型與之對應下面是一些常用的 Channel 類型:

NioSocketChannel, 代表異步的客戶端 TCP Socket 連接.

NioServerSocketChannel, 異步的服務器端 TCP Socket 連接.

NioDatagramChannel, 異步的 UDP 連接

NioSctpChannel, 異步的客戶端 Sctp 連接.

NioSctpServerChannel, 異步的 Sctp 服務器端連接.

OioSocketChannel, 同步的客戶端 TCP Socket 連接.

OioServerSocketChannel, 同步的服務器端 TCP Socket 連接.

OioDatagramChannel, 同步的 UDP 連接

OioSctpChannel, 同步的 Sctp 服務器端連接.

OioSctpServerChannel, 同步的客戶端 TCP Socket 連接.

那么我們是如何設置所需要的 Channel 的類型的呢? 答案是 channel() 方法的調用.
回想一下我們在客戶端連接代碼的初始化 Bootstrap 中, 會調用 channel() 方法, 傳入 NioSocketChannel.class, 這個方法其實就是初始化了一個 BootstrapChannelFactory:

public B channel(Class channelClass) {
    if (channelClass == null) {
        throw new NullPointerException("channelClass");
    }
    return channelFactory(new BootstrapChannelFactory(channelClass));
}

而 BootstrapChannelFactory 實現了 ChannelFactory 接口, 它提供了唯一的方法, 即 newChannel. ChannelFactory, 顧名思義, 就是產生 Channel 的工廠類.
進入到 BootstrapChannelFactory.newChannel 中, 我們看到其實現代碼如下:

@Override
public T newChannel() {
    // 刪除 try 塊
    return clazz.newInstance();
}

根據上面代碼的提示, 我們就可以確定:

Bootstrap 中的 ChannelFactory 的實現是 BootstrapChannelFactory

生成的 Channel 的具體類型是 NioSocketChannel.
Channel 的實例化過程, 其實就是調用的 ChannelFactory#newChannel 方法, 而實例化的 Channel 的具體的類型又是和在初始化 Bootstrap 時傳入的 channel() 方法的參數相關. 因此對于我們這個例子中的客戶端的 Bootstrap 而言, 生成的的 Channel 實例就是 NioSocketChannel.

前面我們已經知道了如何確定一個 Channel 的類型, 并且了解到 Channel 是通過工廠方法 ChannelFactory.newChannel() 來實例化的, 那么 ChannelFactory.newChannel() 方法在哪里調用呢?
繼續跟蹤, 我們發現其調用鏈是:

Bootstrap.connect -> Bootstrap.doConnect -> AbstractBootstrap.initAndRegister

在 AbstractBootstrap.initAndRegister 中就調用了 channelFactory().newChannel() 來獲取一個新的 NioSocketChannel 實例, 其源碼如下:

final ChannelFuture initAndRegister() {
    // 去掉非關鍵代碼
    final Channel channel = channelFactory().newChannel();
    init(channel);
    ChannelFuture regFuture = group().register(channel);
}

在 newChannel 中, 通過類對象的 newInstance 來獲取一個新 Channel 實例, 因而會調用NioSocketChannel 的默認構造器.
NioSocketChannel 默認構造器代碼如下:

public NioSocketChannel() {
    this(newSocket(DEFAULT_SELECTOR_PROVIDER));
}

這里的代碼比較關鍵, 我們看到, 在這個構造器中, 會調用 newSocket 來打開一個新的 Java NIO SocketChannel:

private static SocketChannel newSocket(SelectorProvider provider) {
    ...
    return provider.openSocketChannel();
}

接著會調用父類, 即 AbstractNioByteChannel 的構造器:

AbstractNioByteChannel(Channel parent, SelectableChannel ch)

并傳入參數 parent 為 null, ch 為剛才使用 newSocket 創建的 Java NIO SocketChannel, 因此生成的 NioSocketChannel 的 parent channel 是空的.

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

接著會繼續調用父類 AbstractNioChannel 的構造器, 并傳入了參數 readInterestOp = SelectionKey.OP_READ:

protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
    super(parent);
    this.ch = ch;
    this.readInterestOp = readInterestOp;
    // 省略 try 塊
    // 配置 Java NIO SocketChannel 為非阻塞的.
    ch.configureBlocking(false);
}

然后繼續調用父類 AbstractChannel 的構造器:

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    unsafe = newUnsafe();
    pipeline = new DefaultChannelPipeline(this);
}

到這里, 一個完整的 NioSocketChannel 就初始化完成了, 我們可以稍微總結一下構造一個 NioSocketChannel 所需要做的工作:

調用 NioSocketChannel.newSocket(DEFAULT_SELECTOR_PROVIDER) 打開一個新的 Java NIO SocketChannel

AbstractChannel(Channel parent) 中初始化 AbstractChannel 的屬性:

parent 屬性置為 null

unsafe 通過newUnsafe() 實例化一個 unsafe 對象, 它的類型是 AbstractNioByteChannel.NioByteUnsafe 內部類

pipeline 是 new DefaultChannelPipeline(this) 新創建的實例. 這里體現了:Each channel has its own pipeline and it is created automatically when a new channel is created.

AbstractNioChannel 中的屬性:

SelectableChannel ch 被設置為 Java SocketChannel, 即 NioSocketChannel#newSocket 返回的 Java NIO SocketChannel.

readInterestOp 被設置為 SelectionKey.OP_READ

SelectableChannel ch 被配置為非阻塞的 ch.configureBlocking(false)

NioSocketChannel 中的屬性:

SocketChannelConfig config = new NioSocketChannelConfig(this, socket.socket())

我們簡單地提到了, 在實例化 NioSocketChannel 的過程中, 會在父類 AbstractChannel 的構造器中, 調用 newUnsafe() 來獲取一個 unsafe 實例. 那么 unsafe 是怎么初始化的呢? 它的作用是什么?
其實 unsafe 特別關鍵, 它封裝了對 Java 底層 Socket 的操作, 因此實際上是溝通 Netty 上層和 Java 底層的重要的橋梁.

那么我們就來看一下 Unsafe 接口所提供的方法吧:

interface Unsafe {
    SocketAddress localAddress();
    SocketAddress remoteAddress();
    void register(EventLoop eventLoop, ChannelPromise promise);
    void bind(SocketAddress localAddress, ChannelPromise promise);
    void connect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise);
    void disconnect(ChannelPromise promise);
    void close(ChannelPromise promise);
    void closeForcibly();
    void deregister(ChannelPromise promise);
    void beginRead();
    void write(Object msg, ChannelPromise promise);
    void flush();
    ChannelPromise voidPromise();
    ChannelOutboundBuffer outboundBuffer();
}

一看便知, 這些方法其實都會對應到相關的 Java 底層的 Socket 的操作.
回到 AbstractChannel 的構造方法中, 在這里調用了 newUnsafe() 獲取一個新的 unsafe 對象, 而 newUnsafe 方法在 NioSocketChannel 中被重寫了:

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

NioSocketChannel.newUnsafe 方法會返回一個 NioSocketChannelUnsafe 實例. 從這里我們就可以確定了, 在實例化的 NioSocketChannel 中的 unsafe 字段, 其實是一個 NioSocketChannelUnsafe 的實例.

上面我們分析了一個 Channel (在這個例子中是 NioSocketChannel) 的大體初始化過程, 但是我們漏掉了一個關鍵的部分, 即 ChannelPipeline 的初始化.
根據 Each channel has its own pipeline and it is created automatically when a new channel is created., 我們知道, 在實例化一個 Channel 時, 必然伴隨著實例化一個 ChannelPipeline. 而我們確實在 AbstractChannel 的構造器看到了 pipeline 字段被初始化為 DefaultChannelPipeline 的實例. 那么我們就來看一下, DefaultChannelPipeline 構造器做了哪些工作吧:

public DefaultChannelPipeline(AbstractChannel channel) {
    if (channel == null) {
        throw new NullPointerException("channel");
    }
    this.channel = channel;

    tail = new TailContext(this);
    head = new HeadContext(this);

    head.next = tail;
    tail.prev = head;
}

我們調用 DefaultChannelPipeline 的構造器, 傳入了一個 channel, 而這個 channel 其實就是我們實例化的 NioSocketChannel, DefaultChannelPipeline 會將這個 NioSocketChannel 對象保存在channel 字段中. DefaultChannelPipeline 中, 還有兩個特殊的字段, 即 head 和 tail, 而這兩個字段是一個雙向鏈表的頭和尾. 其實在 DefaultChannelPipeline 中, 維護了一個以 AbstractChannelHandlerContext 為節點的雙向鏈表, 這個鏈表是 Netty 實現 Pipeline 機制的關鍵. 關于 DefaultChannelPipeline 中的雙向鏈表以及它所起的作用, 我在這里暫時不表, 在 Netty 源碼分析之 二 貫穿Netty 的大動脈 ── ChannelPipeline 中會有詳細的分析.

HeadContext 的繼承層次結構如下所示:

TailContext 的繼承層次結構如下所示:

我們可以看到, 鏈表中 head 是一個 ChannelOutboundHandler, 而 tail 則是一個 ChannelInboundHandler.
接著看一下 HeadContext 的構造器:

HeadContext(DefaultChannelPipeline pipeline) {
    super(pipeline, null, HEAD_NAME, false, true);
    unsafe = pipeline.channel().unsafe();
}

它調用了父類 AbstractChannelHandlerContext 的構造器, 并傳入參數 inbound = false, outbound = true.
TailContext 的構造器與 HeadContext 的相反, 它調用了父類 AbstractChannelHandlerContext 的構造器, 并傳入參數 inbound = true, outbound = false.
即 header 是一個 outboundHandler, 而 tail 是一個inboundHandler, 關于這一點, 大家要特別注意, 因為在分析到 Netty Pipeline 時, 我們會反復用到 inbound 和 outbound 這兩個屬性.

回到最開始的 EchoClient.java 代碼中, 我們在一開始就實例化了一個 NioEventLoopGroup 對象, 因此我們就從它的構造器中追蹤一下 EventLoop 的初始化過程.
首先來看一下 NioEventLoopGroup 的類繼承層次:

NioEventLoop 有幾個重載的構造器, 不過內容都沒有什么大的區別, 最終都是調用的父類MultithreadEventLoopGroup構造器:

protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
    super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);
}

其中有一點有意思的地方是, 如果我們傳入的線程數 nThreads 是0, 那么 Netty 會為我們設置默認的線程數 DEFAULT_EVENT_LOOP_THREADS, 而這個默認的線程數是怎么確定的呢?
其實很簡單, 在靜態代碼塊中, 會首先確定 DEFAULT_EVENT_LOOP_THREADS 的值:

static {
    DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
            "io.netty.eventLoopThreads", Runtime.getRuntime().availableProcessors() * 2));
}

Netty 會首先從系統屬性中獲取 "io.netty.eventLoopThreads" 的值, 如果我們沒有設置它的話, 那么就返回默認值: 處理器核心數 * 2.

回到MultithreadEventLoopGroup構造器中, 這個構造器會繼續調用父類 MultithreadEventExecutorGroup 的構造器:

protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
    // 去掉了參數檢查, 異常處理 等代碼.
    children = new SingleThreadEventExecutor[nThreads];
    if (isPowerOfTwo(children.length)) {
        chooser = new PowerOfTwoEventExecutorChooser();
    } else {
        chooser = new GenericEventExecutorChooser();
    }

    for (int i = 0; i < nThreads; i ++) {
        children[i] = newChild(threadFactory, args);
    }
}

根據代碼, 我們就很清楚 MultithreadEventExecutorGroup 中的處理邏輯了:

創建一個大小為 nThreads 的 SingleThreadEventExecutor 數組

根據 nThreads 的大小, 創建不同的 Chooser, 即如果 nThreads 是 2 的冪, 則使用 PowerOfTwoEventExecutorChooser, 反之使用 GenericEventExecutorChooser. 不論使用哪個 Chooser, 它們的功能都是一樣的, 即從 children 數組中選出一個合適的 EventExecutor 實例.

調用 newChhild 方法初始化 children 數組.

根據上面的代碼, 我們知道, MultithreadEventExecutorGroup 內部維護了一個 EventExecutor 數組, Netty 的 EventLoopGroup 的實現機制其實就建立在 MultithreadEventExecutorGroup 之上. 每當 Netty 需要一個 EventLoop 時, 會調用 next() 方法獲取一個可用的 EventLoop.
上面代碼的最后一部分是 newChild 方法, 這個是一個抽象方法, 它的任務是實例化 EventLoop 對象. 我們跟蹤一下它的代碼, 可以發現, 這個方法在 NioEventLoopGroup 類中實現了, 其內容很簡單:

@Override
protected EventExecutor newChild(
        ThreadFactory threadFactory, Object... args) throws Exception {
    return new NioEventLoop(this, threadFactory, (SelectorProvider) args[0]);
}

其實就是實例化一個 NioEventLoop 對象, 然后返回它.

最后總結一下整個 EventLoopGroup 的初始化過程吧:

EventLoopGroup(其實是MultithreadEventExecutorGroup) 內部維護一個類型為 EventExecutor children 數組, 其大小是 nThreads, 這樣就構成了一個線程池

如果我們在實例化 NioEventLoopGroup 時, 如果指定線程池大小, 則 nThreads 就是指定的值, 反之是處理器核心數 * 2

MultithreadEventExecutorGroup 中會調用 newChild 抽象方法來初始化 children 數組

抽象方法 newChild 是在 NioEventLoopGroup 中實現的, 它返回一個 NioEventLoop 實例.

NioEventLoop 屬性:

SelectorProvider provider 屬性: NioEventLoopGroup 構造器中通過 SelectorProvider.provider() 獲取一個 SelectorProvider

Selector selector 屬性: NioEventLoop 構造器中通過調用通過 selector = provider.openSelector() 獲取一個 selector 對象.

在前面的分析中, 我們提到, channel 會在 Bootstrap.initAndRegister 中進行初始化, 但是這個方法還會將初始化好的 Channel 注冊到 EventGroup 中. 接下來我們就來分析一下 Channel 注冊的過程.
回顧一下 AbstractBootstrap.initAndRegister 方法:

final ChannelFuture initAndRegister() {
    // 去掉非關鍵代碼
    final Channel channel = channelFactory().newChannel();
    init(channel);
    ChannelFuture regFuture = group().register(channel);
}

當Channel 初始化后, 會緊接著調用 group().register() 方法來注冊 Channel, 我們繼續跟蹤的話, 會發現其調用鏈如下:
AbstractBootstrap.initAndRegister -> MultithreadEventLoopGroup.register -> SingleThreadEventLoop.register -> AbstractUnsafe.register
通過跟蹤調用鏈, 最終我們發現是調用到了 unsafe 的 register 方法, 那么接下來我們就仔細看一下 AbstractUnsafe.register 方法中到底做了什么:

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    // 省略條件判斷和錯誤處理
    AbstractChannel.this.eventLoop = eventLoop;
    register0(promise);
}

首先, 將 eventLoop 賦值給 Channel 的 eventLoop 屬性, 而我們知道這個 eventLoop 對象其實是 MultithreadEventLoopGroup.next() 方法獲取的, 根據我們前面 關于 EventLoop 初始化 小節中, 我們可以確定 next() 方法返回的 eventLoop 對象是 NioEventLoop 實例.
register 方法接著調用了 register0 方法:

private void register0(ChannelPromise promise) {
    boolean firstRegistration = neverRegistered;
    doRegister();
    neverRegistered = false;
    registered = true;
    safeSetSuccess(promise);
    pipeline.fireChannelRegistered();
    // Only fire a channelActive if the channel has never been registered. This prevents firing
    // multiple channel actives if the channel is deregistered and re-registered.
    if (firstRegistration && isActive()) {
        pipeline.fireChannelActive();
    }
}

register0 又調用了 AbstractNioChannel.doRegister:

@Override
protected void doRegister() throws Exception {
    // 省略錯誤處理
    selectionKey = javaChannel().register(eventLoop().selector, 0, this);
}

javaChannel() 這個方法在前面我們已經知道了, 它返回的是一個 Java NIO SocketChannel, 這里我們將這個 SocketChannel 注冊到與 eventLoop 關聯的 selector 上了.

我們總結一下 Channel 的注冊過程:

首先在 AbstractBootstrap.initAndRegister中, 通過 group().register(channel), 調用 MultithreadEventLoopGroup.register 方法

在MultithreadEventLoopGroup.register 中, 通過 next() 獲取一個可用的 SingleThreadEventLoop, 然后調用它的 register

在 SingleThreadEventLoop.register 中, 通過 channel.unsafe().register(this, promise) 來獲取 channel 的 unsafe() 底層操作對象, 然后調用它的 register.

在 AbstractUnsafe.register 方法中, 調用 register0 方法注冊 Channel

在 AbstractUnsafe.register0 中, 調用 AbstractNioChannel.doRegister 方法

AbstractNioChannel.doRegister 方法通過 javaChannel().register(eventLoop().selector, 0, this) 將 Channel 對應的 Java NIO SockerChannel 注冊到一個 eventLoop 的 Selector 中, 并且將當前 Channel 作為 attachment.

總的來說, Channel 注冊過程所做的工作就是將 Channel 與對應的 EventLoop 關聯, 因此這也體現了, 在 Netty 中, 每個 Channel 都會關聯一個特定的 EventLoop, 并且這個 Channel 中的所有 IO 操作都是在這個 EventLoop 中執行的; 當關聯好 Channel 和 EventLoop 后, 會繼續調用底層的 Java NIO SocketChannel 的 register 方法, 將底層的 Java NIO SocketChannel 注冊到指定的 selector 中. 通過這兩步, 就完成了 Netty Channel 的注冊過程.

Netty 的一個強大和靈活之處就是基于 Pipeline 的自定義 handler 機制. 基于此, 我們可以像添加插件一樣自由組合各種各樣的 handler 來完成業務邏輯. 例如我們需要處理 HTTP 數據, 那么就可以在 pipeline 前添加一個 Http 的編解碼的 Handler, 然后接著添加我們自己的業務邏輯的 handler, 這樣網絡上的數據流就向通過一個管道一樣, 從不同的 handler 中流過并進行編解碼, 最終在到達我們自定義的 handler 中.
既然說到這里, 有些讀者朋友肯定會好奇, 既然這個 pipeline 機制是這么的強大, 那么它是怎么實現的呢? 不過我這里不打算詳細展開 Netty 的 ChannelPipeline 的實現機制(具體的細節會在后續的章節中展示), 我在這一小節中, 從簡單的入手, 展示一下我們自定義的 handler 是如何以及何時添加到 ChannelPipeline 中的.
首先讓我們看一下如下的代碼片段:

...
.handler(new ChannelInitializer() {
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline p = ch.pipeline();
         if (sslCtx != null) {
             p.addLast(sslCtx.newHandler(ch.alloc(), HOST, PORT));
         }
         //p.addLast(new LoggingHandler(LogLevel.INFO));
         p.addLast(new EchoClientHandler());
     }
 });

這個代碼片段就是實現了 handler 的添加功能. 我們看到, Bootstrap.handler 方法接收一個 ChannelHandler, 而我們傳遞的是一個 派生于 ChannelInitializer 的匿名類, 它正好也實現了 ChannelHandler 接口. 我們來看一下, ChannelInitializer 類內到底有什么玄機:

@Sharable
public abstract class ChannelInitializer extends ChannelInboundHandlerAdapter {

    private static final InternalLogger logger = InternalLoggerFactory.getInstance(ChannelInitializer.class);
    protected abstract void initChannel(C ch) throws Exception;

    @Override
    @SuppressWarnings("unchecked")
    public final void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        initChannel((C) ctx.channel());
        ctx.pipeline().remove(this);
        ctx.fireChannelRegistered();
    }
    ...
}

ChannelInitializer 是一個抽象類, 它有一個抽象的方法 initChannel, 我們正是實現了這個方法, 并在這個方法中添加的自定義的 handler 的. 那么 initChannel 是哪里被調用的呢? 答案是 ChannelInitializer.channelRegistered 方法中.
我們來關注一下 channelRegistered 方法. 從上面的源碼中, 我們可以看到, 在 channelRegistered 方法中, 會調用 initChannel 方法, 將自定義的 handler 添加到 ChannelPipeline 中, 然后調用 ctx.pipeline().remove(this) 將自己從 ChannelPipeline 中刪除. 上面的分析過程, 可以用如下圖片展示:
一開始, ChannelPipeline 中只有三個 handler, head, tail 和我們添加的 ChannelInitializer.

接著 initChannel 方法調用后, 添加了自定義的 handler:

最后將 ChannelInitializer 刪除:

分析到這里, 我們已經簡單了解了自定義的 handler 是如何添加到 ChannelPipeline 中的, 不過限于主題與篇幅的原因, 我沒有在這里詳細展開 ChannelPipeline 的底層機制, 我打算在下一篇 Netty 源碼分析之 二 貫穿Netty 的大動脈 ── ChannelPipeline 中對這個問題進行深入的探討.

經過上面的各種分析后, 我們大致了解了 Netty 初始化時, 所做的工作, 那么接下來我們就直奔主題, 分析一下客戶端是如何發起 TCP 連接的.

首先, 客戶端通過調用 Bootstrap 的 connect 方法進行連接.
在 connect 中, 會進行一些參數檢查后, 最終調用的是 doConnect0 方法, 其實現如下:

private static void doConnect0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                if (localAddress == null) {
                    channel.connect(remoteAddress, promise);
                } else {
                    channel.connect(remoteAddress, localAddress, promise);
                }
                promise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

在 doConnect0 中, 會在 event loop 線程中調用 Channel 的 connect 方法, 而這個 Channel 的具體類型是什么呢? 我們在 Channel 初始化這一小節中已經分析過了, 這里 channel 的類型就是 NioSocketChannel.
進行跟蹤到 channel.connect 中, 我們發現它調用的是 DefaultChannelPipeline#connect, 而, pipeline 的 connect 代碼如下:

@Override
public ChannelFuture connect(SocketAddress remoteAddress) {
    return tail.connect(remoteAddress);
}

而 tail 字段, 我們已經分析過了, 是一個 TailContext 的實例, 而 TailContext 又是 AbstractChannelHandlerContext 的子類, 并且沒有實現 connect 方法, 因此這里調用的其實是 AbstractChannelHandlerContext.connect, 我們看一下這個方法的實現:

@Override
public ChannelFuture connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {

    // 刪除的參數檢查的代碼
    final AbstractChannelHandlerContext next = findContextOutbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeConnect(remoteAddress, localAddress, promise);
    } else {
        safeExecute(executor, new OneTimeTask() {
            @Override
            public void run() {
                next.invokeConnect(remoteAddress, localAddress, promise);
            }
        }, promise, null);
    }

    return promise;
}

上面的代碼中有一個關鍵的地方, 即 final AbstractChannelHandlerContext next = findContextOutbound(), 這里調用 findContextOutbound 方法, 從 DefaultChannelPipeline 內的雙向鏈表的 tail 開始, 不斷向前尋找第一個 outbound 為 true 的 AbstractChannelHandlerContext, 然后調用它的 invokeConnect 方法, 其代碼如下:

private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
    // 忽略 try 塊
    ((ChannelOutboundHandler) handler()).connect(this, remoteAddress, localAddress, promise);
}

還記得我們在 "關于 pipeline 的初始化" 這一小節分析的的內容嗎? 我們提到, 在 DefaultChannelPipeline 的構造器中, 會實例化兩個對象: head 和 tail, 并形成了雙向鏈表的頭和尾. head 是 HeadContext 的實例, 它實現了 ChannelOutboundHandler 接口, 并且它的 outbound 字段為 true. 因此在 findContextOutbound 中, 找到的 AbstractChannelHandlerContext 對象其實就是 head. 進而在 invokeConnect 方法中, 我們向上轉換為 ChannelOutboundHandler 就是沒問題的了.
而又因為 HeadContext 重寫了 connect 方法, 因此實際上調用的是 HeadContext.connect. 我們接著跟蹤到 HeadContext.connect, 其代碼如下:

@Override
public void connect(
        ChannelHandlerContext ctx,
        SocketAddress remoteAddress, SocketAddress localAddress,
        ChannelPromise promise) throws Exception {
    unsafe.connect(remoteAddress, localAddress, promise);
}

這個 connect 方法很簡單, 僅僅調用了 unsafe 的 connect 方法. 而 unsafe 又是什么呢?
回顧一下 HeadContext 的構造器, 我們發現 unsafe 是 pipeline.channel().unsafe() 返回的, 而 Channel 的 unsafe 字段, 在這個例子中, 我們已經知道了, 其實是 AbstractNioByteChannel.NioByteUnsafe 內部類. 兜兜轉轉了一大圈, 我們找到了創建 Socket 連接的關鍵代碼.
進行跟蹤 NioByteUnsafe -> AbstractNioUnsafe.connect:

@Override
public final void connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    boolean wasActive = isActive();
    if (doConnect(remoteAddress, localAddress)) {
        fulfillConnectPromise(promise, wasActive);
    } else {
        ...
    }
}

AbstractNioUnsafe.connect 的實現如上代碼所示, 在這個 connect 方法中, 調用了 doConnect 方法, 注意, 這個方法并不是 AbstractNioUnsafe 的方法, 而是 AbstractNioChannel 的抽象方法. doConnect 方法是在 NioSocketChannel 中實現的, 因此進入到 NioSocketChannel.doConnect 中:

@Override
protected boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception {
    if (localAddress != null) {
        javaChannel().socket().bind(localAddress);
    }

    boolean success = false;
    try {
        boolean connected = javaChannel().connect(remoteAddress);
        if (!connected) {
            selectionKey().interestOps(SelectionKey.OP_CONNECT);
        }
        success = true;
        return connected;
    } finally {
        if (!success) {
            doClose();
        }
    }
}

我們終于看到的最關鍵的部分了, 慶祝一下!
上面的代碼不用多說, 首先是獲取 Java NIO SocketChannel, 即我們已經分析過的, 從 NioSocketChannel.newSocket 返回的 SocketChannel 對象; 然后是調用 SocketChannel.connect 方法完成 Java NIO 層面上的 Socket 的連接.
最后, 上面的代碼流程可以用如下時序圖直觀地展示:

本文由 yongshun 發表于個人博客, 采用 署名-相同方式共享 3.0 中國大陸許可協議.
Email: yongshun1228@gmail.com
本文標題為: Netty 源碼分析之 一 揭開 Bootstrap 神秘的紅蓋頭 (客戶端)
本文鏈接為: https://segmentfault.com/a/1190000007282789