摘要：目錄源碼之下無(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 (一)

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

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

我們注意到, pipeline.addXXX 都有一個(gè)重載的方法, 例如 addLast, 它有一個(gè)重載的版本是:

ChannelPipeline addLast(String name, ChannelHandler handler);

第一個(gè)參數(shù)指定了所添加的 handler 的名字(更準(zhǔn)確地說(shuō)是 ChannelHandlerContext 的名字, 不過(guò)我們通常是以 handler 作為敘述的對(duì)象, 因此說(shuō)成 handler 的名字便于理解). 那么 handler 的名字有什么用呢? 如果我們不設(shè)置name, 那么 handler 會(huì)有怎樣的名字?
為了解答這些疑惑, 老規(guī)矩, 依然是從源碼中找到答案.
我們還是以 addLast 方法為例:

@Override
public ChannelPipeline addLast(String name, ChannelHandler handler) {
    return addLast(null, name, handler);
}

這個(gè)方法會(huì)調(diào)用重載的 addLast 方法:

@Override
public ChannelPipeline addLast(EventExecutorGroup group, final String name, ChannelHandler handler) {
    synchronized (this) {
        checkDuplicateName(name);

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

    return this;
}

第一個(gè)參數(shù)被設(shè)置為 null, 我們不關(guān)心它. 第二參數(shù)就是這個(gè) handler 的名字. 看代碼可知, 在添加一個(gè) handler 之前, 需要調(diào)用 checkDuplicateName 方法來(lái)確定此 handler 的名字是否和已添加的 handler 的名字重復(fù). 而這個(gè) checkDuplicateName 方法我們?cè)谇懊嬉呀?jīng)有提到, 這里再回顧一下:

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

Netty 判斷一個(gè) handler 的名字是否重復(fù)的依據(jù)很簡(jiǎn)單: DefaultChannelPipeline 中有一個(gè) 類型為 Map 的 name2ctx 字段, 它的 key 是一個(gè) handler 的名字, 而 value 則是這個(gè) handler 所對(duì)應(yīng)的 ChannelHandlerContext. 每當(dāng)新添加一個(gè) handler 時(shí), 就會(huì) put 到 name2ctx 中. 因此檢查 name2ctx 中是否包含這個(gè) name 即可.
當(dāng)沒(méi)有重名的 handler 時(shí), 就為這個(gè) handler 生成一個(gè)關(guān)聯(lián)的 DefaultChannelHandlerContext 對(duì)象, 然后就將 name 和 newCtx 作為 key-value 對(duì) 放到 name2Ctx 中.

如果我們調(diào)用的是如下的 addLast 方法

ChannelPipeline addLast(ChannelHandler... handlers);

那么 Netty 會(huì)調(diào)用 generateName 為我們的 handler 自動(dòng)生成一個(gè)名字:

private String generateName(ChannelHandler handler) {
    WeakHashMap, String> cache = nameCaches[(int) (Thread.currentThread().getId() % nameCaches.length)];
    Class handlerType = handler.getClass();
    String name;
    synchronized (cache) {
        name = cache.get(handlerType);
        if (name == null) {
            name = generateName0(handlerType);
            cache.put(handlerType, name);
        }
    }

    synchronized (this) {
        // It"s not very likely for a user to put more than one handler of the same type, but make sure to avoid
        // any name conflicts.  Note that we don"t cache the names generated here.
        if (name2ctx.containsKey(name)) {
            String baseName = name.substring(0, name.length() - 1); // Strip the trailing "0".
            for (int i = 1;; i ++) {
                String newName = baseName + i;
                if (!name2ctx.containsKey(newName)) {
                    name = newName;
                    break;
                }
            }
        }
    }

    return name;
}

而 generateName 會(huì)接著調(diào)用 generateName0 來(lái)實(shí)際產(chǎn)生一個(gè) handler 的名字:

private static String generateName0(Class handlerType) {
    return StringUtil.simpleClassName(handlerType) + "#0";
}

自動(dòng)生成的名字的規(guī)則很簡(jiǎn)單, 就是 handler 的簡(jiǎn)單類名加上 "#0", 因此我們的 EchoClientHandler 的名字就是 "EchoClientHandler#0", 這一點(diǎn)也可以通過(guò)調(diào)試窗口佐證:

前面章節(jié)中, 我們知道 AbstractChannelHandlerContext 中有 inbound 和 outbound 兩個(gè) boolean 變量, 分別用于標(biāo)識(shí) Context 所對(duì)應(yīng)的 handler 的類型, 即:

inbound 為真時(shí), 表示對(duì)應(yīng)的 ChannelHandler 實(shí)現(xiàn)了 ChannelInboundHandler 方法.

outbound 為真時(shí), 表示對(duì)應(yīng)的 ChannelHandler 實(shí)現(xiàn)了 ChannelOutboundHandler 方法.

讀者朋友肯定很疑惑了吧: 那究竟這兩個(gè)字段有什么作用呢? 其實(shí)這還要從 ChannelPipeline 的傳輸?shù)氖录愋驼f(shuō)起.
Netty 的事件可以分為 Inbound 和 Outbound 事件.

如下是從 Netty 官網(wǎng)上拷貝的一個(gè)圖示:

                                             I/O Request
                                        via Channel or
                                    ChannelHandlerContext
                                                  |
+---------------------------------------------------+---------------+
|                           ChannelPipeline         |               |
|                                                  |/              |
|    +---------------------+            +-----------+----------+    |
|    | Inbound Handler  N  |            | Outbound Handler  1  |    |
|    +----------+----------+            +-----------+----------+    |
|              /|                                  |               |
|               |                                  |/              |
|    +----------+----------+            +-----------+----------+    |
|    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
|    +----------+----------+            +-----------+----------+    |
|              /|                                  .               |
|               .                                   .               |
| ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
|        [ method call]                       [method call]         |
|               .                                   .               |
|               .                                  |/              |
|    +----------+----------+            +-----------+----------+    |
|    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
|    +----------+----------+            +-----------+----------+    |
|              /|                                  |               |
|               |                                  |/              |
|    +----------+----------+            +-----------+----------+    |
|    | Inbound Handler  1  |            | Outbound Handler  M  |    |
|    +----------+----------+            +-----------+----------+    |
|              /|                                  |               |
+---------------+-----------------------------------+---------------+
              |                                  |/
+---------------+-----------------------------------+---------------+
|               |                                   |               |
|       [ Socket.read() ]                    [ Socket.write() ]     |
|                                                                   |
|  Netty Internal I/O Threads (Transport Implementation)            |
+-------------------------------------------------------------------+

從上圖可以看出, inbound 事件和 outbound 事件的流向是不一樣的, inbound 事件的流行是從下至上, 而 outbound 剛好相反, 是從上到下. 并且 inbound 的傳遞方式是通過(guò)調(diào)用相應(yīng)的 ChannelHandlerContext.fireIN_EVT() 方法, 而 outbound 方法的的傳遞方式是通過(guò)調(diào)用 ChannelHandlerContext.OUT_EVT() 方法. 例如 ChannelHandlerContext.fireChannelRegistered() 調(diào)用會(huì)發(fā)送一個(gè) ChannelRegistered 的 inbound 給下一個(gè)ChannelHandlerContext, 而 ChannelHandlerContext.bind 調(diào)用會(huì)發(fā)送一個(gè) bind 的 outbound 事件給 下一個(gè) ChannelHandlerContext.

Inbound 事件傳播方法有:

ChannelHandlerContext.fireChannelRegistered()
ChannelHandlerContext.fireChannelActive()
ChannelHandlerContext.fireChannelRead(Object)
ChannelHandlerContext.fireChannelReadComplete()
ChannelHandlerContext.fireExceptionCaught(Throwable)
ChannelHandlerContext.fireUserEventTriggered(Object)
ChannelHandlerContext.fireChannelWritabilityChanged()
ChannelHandlerContext.fireChannelInactive()
ChannelHandlerContext.fireChannelUnregistered()

Oubound 事件傳輸方法有:

ChannelHandlerContext.bind(SocketAddress, ChannelPromise)
ChannelHandlerContext.connect(SocketAddress, SocketAddress, ChannelPromise)
ChannelHandlerContext.write(Object, ChannelPromise)
ChannelHandlerContext.flush()
ChannelHandlerContext.read()
ChannelHandlerContext.disconnect(ChannelPromise)
ChannelHandlerContext.close(ChannelPromise)

注意, 如果我們捕獲了一個(gè)事件, 并且想讓這個(gè)事件繼續(xù)傳遞下去, 那么需要調(diào)用 Context 相應(yīng)的傳播方法.
例如:

public class MyInboundHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("Connected!");
        ctx.fireChannelActive();
    }
}

public clas MyOutboundHandler extends ChannelOutboundHandlerAdapter {
    @Override
    public void close(ChannelHandlerContext ctx, ChannelPromise promise) {
        System.out.println("Closing ..");
        ctx.close(promise);
    }
}

上面的例子中, MyInboundHandler 收到了一個(gè) channelActive 事件, 它在處理后, 如果希望將事件繼續(xù)傳播下去, 那么需要接著調(diào)用 ctx.fireChannelActive().

Outbound 事件都是請(qǐng)求事件(request event), 即請(qǐng)求某件事情的發(fā)生, 然后通過(guò) Outbound 事件進(jìn)行通知.
Outbound 事件的傳播方向是 tail -> customContext -> head.

我們接下來(lái)以 connect 事件為例, 分析一下 Outbound 事件的傳播機(jī)制.
首先, 當(dāng)用戶調(diào)用了 Bootstrap.connect 方法時(shí), 就會(huì)觸發(fā)一個(gè) Connect 請(qǐng)求事件, 此調(diào)用會(huì)觸發(fā)如下調(diào)用鏈:

Bootstrap.connect -> Bootstrap.doConnect -> Bootstrap.doConnect0 -> AbstractChannel.connect

繼續(xù)跟蹤的話, 我們就發(fā)現(xiàn), AbstractChannel.connect 其實(shí)由調(diào)用了 DefaultChannelPipeline.connect 方法:

@Override
public ChannelFuture connect(SocketAddress remoteAddress, ChannelPromise promise) {
    return pipeline.connect(remoteAddress, promise);
}

而 pipeline.connect 的實(shí)現(xiàn)如下:

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

可以看到, 當(dāng) outbound 事件(這里是 connect 事件)傳遞到 Pipeline 后, 它其實(shí)是以 tail 為起點(diǎn)開(kāi)始傳播的.
而 tail.connect 其實(shí)調(diào)用的是 AbstractChannelHandlerContext.connect 方法:

@Override
public ChannelFuture connect(
        final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
    ...
    final AbstractChannelHandlerContext next = findContextOutbound();
    EventExecutor executor = next.executor();
    ...
    next.invokeConnect(remoteAddress, localAddress, promise);
    ...
    return promise;
}

findContextOutbound() 顧名思義, 它的作用是以當(dāng)前 Context 為起點(diǎn), 向 Pipeline 中的 Context 雙向鏈表的前端尋找第一個(gè) outbound 屬性為真的 Context(即關(guān)聯(lián)著 ChannelOutboundHandler 的 Context), 然后返回.
它的實(shí)現(xiàn)如下:

private AbstractChannelHandlerContext findContextOutbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.prev;
    } while (!ctx.outbound);
    return ctx;
}

當(dāng)我們找到了一個(gè) outbound 的 Context 后, 就調(diào)用它的 invokeConnect 方法, 這個(gè)方法中會(huì)調(diào)用 Context 所關(guān)聯(lián)著的 ChannelHandler 的 connect 方法:

private void invokeConnect(SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
    try {
        ((ChannelOutboundHandler) handler()).connect(this, remoteAddress, localAddress, promise);
    } catch (Throwable t) {
        notifyOutboundHandlerException(t, promise);
    }
}

如果用戶沒(méi)有重寫 ChannelHandler 的 connect 方法, 那么會(huì)調(diào)用 ChannelOutboundHandlerAdapter 所實(shí)現(xiàn)的方法:

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

我們看到, ChannelOutboundHandlerAdapter.connect 僅僅調(diào)用了 ctx.connect, 而這個(gè)調(diào)用又回到了:

Context.connect -> Connect.findContextOutbound -> next.invokeConnect -> handler.connect -> Context.connect

這樣的循環(huán)中, 直到 connect 事件傳遞到DefaultChannelPipeline 的雙向鏈表的頭節(jié)點(diǎn), 即 head 中. 為什么會(huì)傳遞到 head 中呢? 回想一下, head 實(shí)現(xiàn)了 ChannelOutboundHandler, 因此它的 outbound 屬性是 true.
因?yàn)?head 本身既是一個(gè) ChannelHandlerContext, 又實(shí)現(xiàn)了 ChannelOutboundHandler 接口, 因此當(dāng) connect 消息傳遞到 head 后, 會(huì)將消息轉(zhuǎn)遞到對(duì)應(yīng)的 ChannelHandler 中處理, 而恰好, head 的 handler() 返回的就是 head 本身:

@Override
public ChannelHandler handler() {
    return this;
}

因此最終 connect 事件是在 head 中處理的. head 的 connect 事件處理方法如下:

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

到這里, 整個(gè) Connect 請(qǐng)求事件就結(jié)束了.
下面以一幅圖來(lái)描述一個(gè)整個(gè) Connect 請(qǐng)求事件的處理過(guò)程:

點(diǎn)此有無(wú)碼高清原圖

我們僅僅以 Connect 請(qǐng)求事件為例, 分析了 Outbound 事件的傳播過(guò)程, 但是其實(shí)所有的 outbound 的事件傳播都遵循著一樣的傳播規(guī)律, 讀者可以試著分析一下其他的 outbound 事件, 體會(huì)一下它們的傳播過(guò)程.

Inbound 事件和 Outbound 事件的處理過(guò)程有點(diǎn)鏡像.
Inbound 事件是一個(gè)通知事件, 即某件事已經(jīng)發(fā)生了, 然后通過(guò) Inbound 事件進(jìn)行通知. Inbound 通常發(fā)生在 Channel 的狀態(tài)的改變或 IO 事件就緒.
Inbound 的特點(diǎn)是它傳播方向是 head -> customContext -> tail.

既然上面我們分析了 Connect 這個(gè) Outbound 事件, 那么接著分析 Connect 事件后會(huì)發(fā)生什么 Inbound 事件, 并最終找到 Outbound 和 Inbound 事件之間的聯(lián)系.

當(dāng) Connect 這個(gè) Outbound 傳播到 unsafe 后, 其實(shí)是在 AbstractNioUnsafe.connect 方法中進(jìn)行處理的:

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

在 AbstractNioUnsafe.connect 中, 首先調(diào)用 doConnect 方法進(jìn)行實(shí)際上的 Socket 連接, 當(dāng)連接上后, 會(huì)調(diào)用 fulfillConnectPromise 方法:

private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
    ...
    // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
    // because what happened is what happened.
    if (!wasActive && isActive()) {
        pipeline().fireChannelActive();
    }
    ...
}

我們看到, 在 fulfillConnectPromise 中, 會(huì)通過(guò)調(diào)用 pipeline().fireChannelActive() 將通道激活的消息(即 Socket 連接成功)發(fā)送出去.
而這里, 當(dāng)調(diào)用 pipeline.fireXXX 后, 就是 Inbound 事件的起點(diǎn).
因此當(dāng)調(diào)用了 pipeline().fireChannelActive() 后, 就產(chǎn)生了一個(gè) ChannelActive Inbound 事件, 我們就從這里開(kāi)始看看這個(gè) Inbound 事件是怎么傳播的吧.

@Override
public ChannelPipeline fireChannelActive() {
    head.fireChannelActive();

    if (channel.config().isAutoRead()) {
        channel.read();
    }

    return this;
}

哈哈, 果然, 在 fireChannelActive 方法中, 調(diào)用的是 head.fireChannelActive, 因此可以證明了, Inbound 事件在 Pipeline 中傳輸?shù)钠瘘c(diǎn)是 head.
那么, 在 head.fireChannelActive() 中又做了什么呢?

@Override
public ChannelHandlerContext fireChannelActive() {
    final AbstractChannelHandlerContext next = findContextInbound();
    EventExecutor executor = next.executor();
    ...
    next.invokeChannelActive();
    ...
    return this;
}

上面的代碼應(yīng)該很熟悉了吧. 回想一下在 Outbound 事件(例如 Connect 事件)的傳輸過(guò)程中時(shí), 我們也有類似的操作:

首先調(diào)用 findContextInbound, 從 Pipeline 的雙向鏈表中中找到第一個(gè)屬性 inbound 為真的 Context, 然后返回

調(diào)用這個(gè) Context 的 invokeChannelActive

invokeChannelActive 方法如下:

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

這個(gè)方法和 Outbound 的對(duì)應(yīng)方法(例如 invokeConnect) 如出一轍. 同 Outbound 一樣, 如果用戶沒(méi)有重寫 channelActive 方法, 那么會(huì)調(diào)用 ChannelInboundHandlerAdapter 的 channelActive 方法:

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
    ctx.fireChannelActive();
}

同樣地, 在 ChannelInboundHandlerAdapter.channelActive 中, 僅僅調(diào)用了 ctx.fireChannelActive 方法, 因此就會(huì)有如下循環(huán):

Context.fireChannelActive -> Connect.findContextInbound -> nextContext.invokeChannelActive -> nextHandler.channelActive -> nextContext.fireChannelActive

這樣的循環(huán)中. 同理, tail 本身 既實(shí)現(xiàn)了 ChannelInboundHandler 接口, 又實(shí)現(xiàn)了 ChannelHandlerContext 接口, 因此當(dāng) channelActive 消息傳遞到 tail 后, 會(huì)將消息轉(zhuǎn)遞到對(duì)應(yīng)的 ChannelHandler 中處理, 而恰好, tail 的 handler() 返回的就是 tail 本身:

@Override
public ChannelHandler handler() {
    return this;
}

因此 channelActive Inbound 事件最終是在 tail 中處理的, 我們看一下它的處理方法:

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception { }

TailContext.channelActive 方法是空的. 如果讀者自行查看 TailContext 的 Inbound 處理方法時(shí), 會(huì)發(fā)現(xiàn), 它們的實(shí)現(xiàn)都是空的. 可見(jiàn), 如果是 Inbound, 當(dāng)用戶沒(méi)有實(shí)現(xiàn)自定義的處理器時(shí), 那么默認(rèn)是不處理的.

用一幅圖來(lái)總結(jié)一下 Inbound 的傳輸過(guò)程吧:

點(diǎn)擊此可以看高清無(wú)碼原圖

對(duì)于 Outbound事件:

Outbound 事件是請(qǐng)求事件(由 Connect 發(fā)起一個(gè)請(qǐng)求, 并最終由 unsafe 處理這個(gè)請(qǐng)求)

Outbound 事件的發(fā)起者是 Channel

Outbound 事件的處理者是 unsafe

Outbound 事件在 Pipeline 中的傳輸方向是 tail -> head.

在 ChannelHandler 中處理事件時(shí), 如果這個(gè) Handler 不是最后一個(gè) Hnalder, 則需要調(diào)用 ctx.xxx (例如 ctx.connect) 將此事件繼續(xù)傳播下去. 如果不這樣做, 那么此事件的傳播會(huì)提前終止.

Outbound 事件流: Context.OUT_EVT -> Connect.findContextOutbound -> nextContext.invokeOUT_EVT -> nextHandler.OUT_EVT -> nextContext.OUT_EVT

對(duì)于 Inbound 事件:

Inbound 事件是通知事件, 當(dāng)某件事情已經(jīng)就緒后, 通知上層.

Inbound 事件發(fā)起者是 unsafe

Inbound 事件的處理者是 Channel, 如果用戶沒(méi)有實(shí)現(xiàn)自定義的處理方法, 那么Inbound 事件默認(rèn)的處理者是 TailContext, 并且其處理方法是空實(shí)現(xiàn).

Inbound 事件在 Pipeline 中傳輸方向是 head -> tail

在 ChannelHandler 中處理事件時(shí), 如果這個(gè) Handler 不是最后一個(gè) Hnalder, 則需要調(diào)用 ctx.fireIN_EVT (例如 ctx.fireChannelActive) 將此事件繼續(xù)傳播下去. 如果不這樣做, 那么此事件的傳播會(huì)提前終止.

Outbound 事件流: Context.fireIN_EVT -> Connect.findContextInbound -> nextContext.invokeIN_EVT -> nextHandler.IN_EVT -> nextContext.fireIN_EVT

outbound 和 inbound 事件十分的鏡像, 并且 Context 與 Handler 直接的調(diào)用關(guān)系是否容易混淆, 因此讀者在閱讀這里的源碼時(shí), 需要特別的注意.

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