摘要：目錄源碼之下無(wú)秘密做最好的源碼分析教程源碼分析之番外篇的前生今世的前生今世之一簡(jiǎn)介的前生今世之二小結(jié)的前生今世之三詳解的前生今世之四詳解源碼分析之零磨刀不誤砍柴工源碼分析環(huán)境搭建源碼分析之一揭開(kāi)神秘的紅蓋頭源碼分析之一揭開(kāi)神秘的紅蓋頭客戶端

源碼之下無(wú)秘密 ── 做最好的 Netty 源碼分析教程

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

Java NIO 的前生今世 之一 簡(jiǎn)介

Java NIO 的前生今世 之二 NIO Channel 小結(jié)

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

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

Netty 源碼分析之 零 磨刀不誤砍柴工 源碼分析環(huán)境搭建

Netty 源碼分析之 一 揭開(kāi) Bootstrap 神秘的紅蓋頭

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

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

Netty 源碼分析之 二 貫穿 Netty 的大動(dòng)脈 ── ChannelPipeline (一)

此文章已同步發(fā)送到我的 github 上

這篇是 Netty 源碼分析 的第二篇, 在這篇文章中, 我會(huì)為讀者詳細(xì)地分析 Netty 中的 ChannelPipeline 機(jī)制.

相信大家都知道了, 在 Netty 中每個(gè) Channel 都有且僅有一個(gè) ChannelPipeline 與之對(duì)應(yīng), 它們的組成關(guān)系如下:

通過(guò)上圖我們可以看到, 一個(gè) Channel 包含了一個(gè) ChannelPipeline, 而 ChannelPipeline 中又維護(hù)了一個(gè)由 ChannelHandlerContext 組成的雙向鏈表. 這個(gè)鏈表的頭是 HeadContext, 鏈表的尾是 TailContext, 并且每個(gè) ChannelHandlerContext 中又關(guān)聯(lián)著一個(gè) ChannelHandler.
上面的圖示給了我們一個(gè)對(duì) ChannelPipeline 的直觀認(rèn)識(shí), 但是實(shí)際上 Netty 實(shí)現(xiàn)的 Channel 是否真的是這樣的呢? 我們繼續(xù)用源碼說(shuō)話.

在第一章 Netty 源碼分析之 一 揭開(kāi) Bootstrap 神秘的紅蓋頭 中, 我們已經(jīng)知道了一個(gè) Channel 的初始化的基本過(guò)程, 下面我們?cè)倩仡櫼幌?
下面的代碼是 AbstractChannel 構(gòu)造器:

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

AbstractChannel 有一個(gè) pipeline 字段, 在構(gòu)造器中會(huì)初始化它為 DefaultChannelPipeline的實(shí)例. 這里的代碼就印證了一點(diǎn): 每個(gè) Channel 都有一個(gè) ChannelPipeline.
接著我們跟蹤一下 DefaultChannelPipeline 的初始化過(guò)程.
首先進(jìn)入到 DefaultChannelPipeline 構(gòu)造器中:

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 構(gòu)造器中, 首先將與之關(guān)聯(lián)的 Channel 保存到字段 channel 中, 然后實(shí)例化兩個(gè) ChannelHandlerContext, 一個(gè)是 HeadContext 實(shí)例 head, 另一個(gè)是 TailContext 實(shí)例 tail. 接著將 head 和 tail 互相指向, 構(gòu)成一個(gè)雙向鏈表.
特別注意到, 我們?cè)陂_(kāi)始的示意圖中, head 和 tail 并沒(méi)有包含 ChannelHandler, 這是因?yàn)?HeadContext 和 TailContext 繼承于 AbstractChannelHandlerContext 的同時(shí)也實(shí)現(xiàn)了 ChannelHandler 接口了, 因此它們有 Context 和 Handler 的雙重屬性.

在第一章的時(shí)候, 我們已經(jīng)對(duì) ChannelPipeline 的初始化有了一個(gè)大致的了解, 不過(guò)當(dāng)時(shí)重點(diǎn)畢竟不在 ChannelPipeline 這里, 因此沒(méi)有深入地分析它的初始化過(guò)程. 那么下面我們就來(lái)看一下具體的 ChannelPipeline 的初始化都做了哪些工作吧.

我們?cè)賮?lái)回顧一下, 在實(shí)例化一個(gè) Channel 時(shí), 會(huì)伴隨著一個(gè) ChannelPipeline 的實(shí)例化, 并且此 Channel 會(huì)與這個(gè) ChannelPipeline 相互關(guān)聯(lián), 這一點(diǎn)可以通過(guò)NioSocketChannel 的父類 AbstractChannel 的構(gòu)造器予以佐證:

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

當(dāng)實(shí)例化一個(gè) Channel(這里以 EchoClient 為例, 那么 Channel 就是 NioSocketChannel), 其 pipeline 字段就是我們新創(chuàng)建的 DefaultChannelPipeline 對(duì)象, 那么我們就來(lái)看一下 DefaultChannelPipeline 的構(gòu)造方法吧:

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 的構(gòu)造方法中, 將傳入的 channel 賦值給字段 this.channel, 接著又實(shí)例化了兩個(gè)特殊的字段: tail 與 head. 這兩個(gè)字段是一個(gè)雙向鏈表的頭和尾. 其實(shí)在 DefaultChannelPipeline 中, 維護(hù)了一個(gè)以 AbstractChannelHandlerContext 為節(jié)點(diǎn)的雙向鏈表, 這個(gè)鏈表是 Netty 實(shí)現(xiàn) Pipeline 機(jī)制的關(guān)鍵.
再回顧一下 head 和 tail 的類層次結(jié)構(gòu):

從類層次結(jié)構(gòu)圖中可以很清楚地看到, head 實(shí)現(xiàn)了 ChannelInboundHandler, 而 tail 實(shí)現(xiàn)了 ChannelOutboundHandler 接口, 并且它們都實(shí)現(xiàn)了 ChannelHandlerContext 接口, 因此可以說(shuō) head 和 tail 即是一個(gè) ChannelHandler, 又是一個(gè) ChannelHandlerContext.

接著看一下 HeadContext 的構(gòu)造器:

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

它調(diào)用了父類 AbstractChannelHandlerContext 的構(gòu)造器, 并傳入?yún)?shù) inbound = false, outbound = true.
TailContext 的構(gòu)造器與 HeadContext 的相反, 它調(diào)用了父類 AbstractChannelHandlerContext 的構(gòu)造器, 并傳入?yún)?shù) inbound = true, outbound = false.
即 header 是一個(gè) outboundHandler, 而 tail 是一個(gè)inboundHandler, 關(guān)于這一點(diǎn), 大家要特別注意, 因?yàn)樵诤竺娴姆治鲋? 我們會(huì)反復(fù)用到 inbound 和 outbound 這兩個(gè)屬性.

上面一小節(jié)中, 我們已經(jīng)分析了 Channel 的組成, 其中我們了解到, 最開(kāi)始的時(shí)候 ChannelPipeline 中含有兩個(gè) ChannelHandlerContext(同時(shí)也是 ChannelHandler), 但是這個(gè) Pipeline并不能實(shí)現(xiàn)什么特殊的功能, 因?yàn)槲覀冞€沒(méi)有給它添加自定義的 ChannelHandler.
通常來(lái)說(shuō), 我們?cè)诔跏蓟?Bootstrap, 會(huì)添加我們自定義的 ChannelHandler, 就以我們熟悉的 EchoClient 來(lái)舉例吧:

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

上面代碼的初始化過(guò)程, 相信大家都不陌生. 在調(diào)用 handler 時(shí), 傳入了 ChannelInitializer 對(duì)象, 它提供了一個(gè) initChannel 方法供我們初始化 ChannelHandler. 那么這個(gè)初始化過(guò)程是怎樣的呢? 下面我們就來(lái)揭開(kāi)它的神秘面紗.

ChannelInitializer 實(shí)現(xiàn)了 ChannelHandler, 那么它是在什么時(shí)候添加到 ChannelPipeline 中的呢? 進(jìn)行了一番搜索后, 我們發(fā)現(xiàn)它是在 Bootstrap.init 方法中添加到 ChannelPipeline 中的.
其代碼如下:

@Override
@SuppressWarnings("unchecked")
void init(Channel channel) throws Exception {
    ChannelPipeline p = channel.pipeline();
    p.addLast(handler());
    ...
}

上面的代碼將 handler() 返回的 ChannelHandler 添加到 Pipeline 中, 而 handler() 返回的是handler 其實(shí)就是我們?cè)诔跏蓟?Bootstrap 調(diào)用 handler 設(shè)置的 ChannelInitializer 實(shí)例, 因此這里就是將 ChannelInitializer 插入到了 Pipeline 的末端.
此時(shí) Pipeline 的結(jié)構(gòu)如下圖所示:

有朋友可能就有疑惑了, 我明明插入的是一個(gè) ChannelInitializer 實(shí)例, 為什么在 ChannelPipeline 中的雙向鏈表中的元素卻是一個(gè) ChannelHandlerContext? 為了解答這個(gè)問(wèn)題, 我們繼續(xù)在代碼中尋找答案吧.
我們剛才提到, 在 Bootstrap.init 中會(huì)調(diào)用 p.addLast() 方法, 將 ChannelInitializer 插入到鏈表末端:

@Override
public ChannelPipeline addLast(EventExecutorGroup group, final String name, ChannelHandler handler) {
    synchronized (this) {
        checkDuplicateName(name); // 檢查此 handler 是否有重復(fù)的名字

        AbstractChannelHandlerContext newCtx = new DefaultChannelHandlerContext(this, group, name, handler);
        addLast0(name, newCtx);
    }

    return this;
}

addLast 有很多重載的方法, 我們關(guān)注這個(gè)比較重要的方法就可以了.
上面的 addLast 方法中, 首先檢查這個(gè) ChannelHandler 的名字是否是重復(fù)的, 如果不重復(fù)的話, 則為這個(gè) Handler 創(chuàng)建一個(gè)對(duì)應(yīng)的 DefaultChannelHandlerContext 實(shí)例, 并與之關(guān)聯(lián)起來(lái)(Context 中有一個(gè) handler 屬性保存著對(duì)應(yīng)的 Handler 實(shí)例). 判斷此 Handler 是否重名的方法很簡(jiǎn)單: Netty 中有一個(gè) name2ctx Map 字段, key 是 handler 的名字, 而 value 則是 handler 本身. 因此通過(guò)如下代碼就可以判斷一個(gè) handler 是否重名了:

private void checkDuplicateName(String name) {
    if (name2ctx.containsKey(name)) {
        throw new IllegalArgumentException("Duplicate handler name: " + name);
    }
}

為了添加一個(gè) handler 到 pipeline 中, 必須把此 handler 包裝成 ChannelHandlerContext. 因此在上面的代碼中我們可以看到新實(shí)例化了一個(gè) newCtx 對(duì)象, 并將 handler 作為參數(shù)傳遞到構(gòu)造方法中. 那么我們來(lái)看一下實(shí)例化的 DefaultChannelHandlerContext 到底有什么玄機(jī)吧.
首先看它的構(gòu)造器:

DefaultChannelHandlerContext(
        DefaultChannelPipeline pipeline, EventExecutorGroup group, String name, ChannelHandler handler) {
    super(pipeline, group, name, isInbound(handler), isOutbound(handler));
    if (handler == null) {
        throw new NullPointerException("handler");
    }
    this.handler = handler;
}

DefaultChannelHandlerContext 的構(gòu)造器中, 調(diào)用了兩個(gè)很有意思的方法: isInbound 與 isOutbound, 這兩個(gè)方法是做什么的呢?

private static boolean isInbound(ChannelHandler handler) {
    return handler instanceof ChannelInboundHandler;
}

private static boolean isOutbound(ChannelHandler handler) {
    return handler instanceof ChannelOutboundHandler;
}

從源碼中可以看到, 當(dāng)一個(gè) handler 實(shí)現(xiàn)了 ChannelInboundHandler 接口, 則 isInbound 返回真; 相似地, 當(dāng)一個(gè) handler 實(shí)現(xiàn)了 ChannelOutboundHandler 接口, 則 isOutbound 就返回真.
而這兩個(gè) boolean 變量會(huì)傳遞到父類 AbstractChannelHandlerContext 中, 并初始化父類的兩個(gè)字段: inbound 與 outbound.
那么這里的 ChannelInitializer 所對(duì)應(yīng)的 DefaultChannelHandlerContext 的 inbound 與 inbound 字段分別是什么呢? 那就看一下 ChannelInitializer 到底實(shí)現(xiàn)了哪個(gè)接口不就行了? 如下是 ChannelInitializer 的類層次結(jié)構(gòu)圖:

可以清楚地看到, ChannelInitializer 僅僅實(shí)現(xiàn)了 ChannelInboundHandler 接口, 因此這里實(shí)例化的 DefaultChannelHandlerContext 的 inbound = true, outbound = false.
不就是 inbound 和 outbound 兩個(gè)字段嘛, 為什么需要這么大費(fèi)周章地分析一番? 其實(shí)這兩個(gè)字段關(guān)系到 pipeline 的事件的流向與分類, 因此是十分關(guān)鍵的, 不過(guò)我在這里先賣個(gè)關(guān)子, 后面我們?cè)賮?lái)詳細(xì)分析這兩個(gè)字段所起的作用. 在這里, 讀者只需要記住, ChannelInitializer 所對(duì)應(yīng)的 DefaultChannelHandlerContext 的 inbound = true, outbound = false 即可.

當(dāng)創(chuàng)建好 Context 后, 就將這個(gè) Context 插入到 Pipeline 的雙向鏈表中:

private void addLast0(final String name, AbstractChannelHandlerContext newCtx) {
    checkMultiplicity(newCtx);

    AbstractChannelHandlerContext prev = tail.prev;
    newCtx.prev = prev;
    newCtx.next = tail;
    prev.next = newCtx;
    tail.prev = newCtx;

    name2ctx.put(name, newCtx);

    callHandlerAdded(newCtx);
}

顯然, 這個(gè)代碼就是典型的雙向鏈表的插入操作了. 當(dāng)調(diào)用了 addLast 方法后, Netty 就會(huì)將此 handler 添加到雙向鏈表中 tail 元素之前的位置.

在上一小節(jié)中, 我們已經(jīng)分析了一個(gè) ChannelInitializer 如何插入到 Pipeline 中的, 接下來(lái)就來(lái)探討一下 ChannelInitializer 在哪里被調(diào)用, ChannelInitializer 的作用, 以及我們自定義的 ChannelHandler 是如何插入到 Pipeline 中的.

在 Netty 源碼分析之 一 揭開(kāi) Bootstrap 神秘的紅蓋頭 (客戶端) 一章的 channel 的注冊(cè)過(guò)程 小節(jié)中, 我們已經(jīng)分析過(guò) Channel 的注冊(cè)過(guò)程了, 這里我們?cè)俸?jiǎn)單地復(fù)習(xí)一下:

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

在MultithreadEventLoopGroup.register 中, 通過(guò) next() 獲取一個(gè)可用的 SingleThreadEventLoop, 然后調(diào)用它的 register

在 SingleThreadEventLoop.register 中, 通過(guò) channel.unsafe().register(this, promise) 來(lái)獲取 channel 的 unsafe() 底層操作對(duì)象, 然后調(diào)用它的 register.

在 AbstractUnsafe.register 方法中, 調(diào)用 register0 方法注冊(cè) Channel

在 AbstractUnsafe.register0 中, 調(diào)用 AbstractNioChannel#doRegister 方法

AbstractNioChannel.doRegister 方法通過(guò) javaChannel().register(eventLoop().selector, 0, this) 將 Channel 對(duì)應(yīng)的 Java NIO SockerChannel 注冊(cè)到一個(gè) eventLoop 的 Selector 中, 并且將當(dāng)前 Channel 作為 attachment.

而我們自定義 ChannelHandler 的添加過(guò)程, 發(fā)生在 AbstractUnsafe.register0 中, 在這個(gè)方法中調(diào)用了 pipeline.fireChannelRegistered() 方法, 其實(shí)現(xiàn)如下:

@Override
public ChannelPipeline fireChannelRegistered() {
    head.fireChannelRegistered();
    return this;
}

上面的代碼很簡(jiǎn)單, 就是調(diào)用了 head.fireChannelRegistered() 方法而已.

關(guān)于上面代碼的 head.fireXXX 的調(diào)用形式, 是 Netty 中 Pipeline 傳遞事件的常用方式, 我們以后會(huì)經(jīng)常看到.

還記得 head 的 類層次結(jié)構(gòu)圖不, head 是一個(gè) AbstractChannelHandlerContext 實(shí)例, 并且它沒(méi)有重寫(xiě) fireChannelRegistered 方法, 因此 head.fireChannelRegistered 其實(shí)是調(diào)用的 AbstractChannelHandlerContext.fireChannelRegistered:

@Override
public ChannelHandlerContext fireChannelRegistered() {
    final AbstractChannelHandlerContext next = findContextInbound();
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRegistered();
    } else {
        executor.execute(new OneTimeTask() {
            @Override
            public void run() {
                next.invokeChannelRegistered();
            }
        });
    }
    return this;
}

這個(gè)方法的第一句是調(diào)用 findContextInbound 獲取一個(gè) Context, 那么它返回的 Context 到底是什么呢? 我們跟進(jìn)代碼中看一下:

private AbstractChannelHandlerContext findContextInbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while (!ctx.inbound);
    return ctx;
}

很顯然, 這個(gè)代碼會(huì)從 head 開(kāi)始遍歷 Pipeline 的雙向鏈表, 然后找到第一個(gè)屬性 inbound 為 true 的 ChannelHandlerContext 實(shí)例. 想起來(lái)了沒(méi)? 我們?cè)谇懊娣治?ChannelInitializer 時(shí), 花了大量的筆墨來(lái)分析了 inbound 和 outbound 屬性, 你看現(xiàn)在這里就用上了. 回想一下, ChannelInitializer 實(shí)現(xiàn)了 ChannelInboudHandler, 因此它所對(duì)應(yīng)的 ChannelHandlerContext 的 inbound 屬性就是 true, 因此這里返回就是 ChannelInitializer 實(shí)例所對(duì)應(yīng)的 ChannelHandlerContext. 即:

當(dāng)獲取到 inbound 的 Context 后, 就調(diào)用它的 invokeChannelRegistered 方法:

private void invokeChannelRegistered() {
    try {
        ((ChannelInboundHandler) handler()).channelRegistered(this);
    } catch (Throwable t) {
        notifyHandlerException(t);
    }
}

我們已經(jīng)強(qiáng)調(diào)過(guò)了, 每個(gè) ChannelHandler 都與一個(gè) ChannelHandlerContext 關(guān)聯(lián), 我們可以通過(guò) ChannelHandlerContext 獲取到對(duì)應(yīng)的 ChannelHandler. 因此很顯然了, 這里 handler() 返回的, 其實(shí)就是一開(kāi)始我們實(shí)例化的 ChannelInitializer 對(duì)象, 并接著調(diào)用了 ChannelInitializer.channelRegistered 方法. 看到這里, 讀者朋友是否會(huì)覺(jué)得有點(diǎn)眼熟呢? ChannelInitializer.channelRegistered 這個(gè)方法我們?cè)诘谝徽碌臅r(shí)候已經(jīng)大量地接觸了, 但是我們并沒(méi)有深入地分析這個(gè)方法的調(diào)用過(guò)程, 那么在這里讀者朋友應(yīng)該對(duì)它的調(diào)用有了更加深入的了解了吧.

那么這個(gè)方法中又有什么玄機(jī)呢? 繼續(xù)看代碼:

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

initChannel 這個(gè)方法我們很熟悉了吧, 它就是我們?cè)诔跏蓟?Bootstrap 時(shí), 調(diào)用 handler 方法傳入的匿名內(nèi)部類所實(shí)現(xiàn)的方法:

.handler(new ChannelInitializer() {
     @Override
     public void initChannel(SocketChannel ch) throws Exception {
         ChannelPipeline p = ch.pipeline();
         p.addLast(new EchoClientHandler());
     }
 });

因此當(dāng)調(diào)用了這個(gè)方法后, 我們自定義的 ChannelHandler 就插入到 Pipeline 了, 此時(shí)的 Pipeline 如下圖所示:

當(dāng)添加了自定義的 ChannelHandler 后, 會(huì)刪除 ChannelInitializer 這個(gè) ChannelHandler, 即 "ctx.pipeline().remove(this)", 因此最后的 Pipeline 如下:

好了, 到了這里, 我們的 自定義 ChannelHandler 的添加過(guò)程 也分析的查不多了.

下一小節(jié) Netty 源碼分析之 二 貫穿Netty 的大動(dòng)脈 ── ChannelPipeline (二)

本文由 yongshun 發(fā)表于個(gè)人博客, 采用 署名-相同方式共享 3.0 中國(guó)大陸許可協(xié)議.
Email: yongshun1228@gmail.com
本文標(biāo)題為: Netty 源碼分析之 二 貫穿Netty 的大動(dòng)脈 ── ChannelPipeline (一)
本文鏈接為: https://segmentfault.com/a/1190000007308934