Netty原始碼分析第四章: pipeline

第二節: Handler的新增

新增handler, 我們以使用者程式碼為例進行剖析:

.childHandler(new ChannelInitializer<SocketChannel>() {
    protected void initChannel(SocketChannel ch) throws Exception {
        ch.pipeline().addLast(new DelimiterBasedFrameDecoder(Integer.MAX_VALUE, Delimiters.lineDelimiter()[0]));
        ch.pipeline().addLast(
 
new StringEncoder());
        ch.pipeline().addLast(new SimpleHandler());
    }
});

用過netty的小夥伴們肯定對這段程式碼不會陌生, 通過addLast, 可以新增編解碼器和我們自定義的handler, 某一個事件完成之後可以自動呼叫我們handler預先定義的方法, 具體新增和呼叫是怎麼個執行邏輯, 在我們之後的內容會全部學習到, 以後再使用這類的功能會得心應手

在這裡, 我們主要剖析 ch.pipeline().addLast(new SimpleHandler()) 這部分程式碼的addLast()方法

首先通過channel拿到當前的pipline, 這個上一小節進行剖析過相信不會陌生

拿到pipeline之後再為其新增handler, 因為channel初始化預設建立的是DefualtChannelPipeline

我們跟到其addLast()方法中:

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

首先看到這裡的引數其實是一個可變物件, 也就是可以傳遞多個handler, 這裡我們只傳遞了一個

我們繼續跟addLast:

public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
    if (handlers == null) {
        throw new NullPointerException("handlers");
    }
    //傳多個引數的時候通過for迴圈新增
    for (ChannelHandler h: handlers) {
        if (h == null) {
            break;
        }
        addLast(executor, null, h);
    }
    return this;
}

這裡如果傳入多個handler則會迴圈新增, 我們通常只新增一個

再繼續跟到addLast()方法中去:

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        //判斷handler是否被重複新增(1)
        checkMultiplicity(handler);
        //建立一個HandlerContext並新增到列表(2)
        newCtx = newContext(group, filterName(name, handler), handler);

        //新增HandlerContext(3)
        addLast0(newCtx);

        //是否已註冊
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            //回撥使用者事件
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    //回撥新增事件(4)
    callHandlerAdded0(newCtx);
    return this;
}

這部分程式碼比較長, 我們拆解為4個步驟:

1.重複新增驗證

2.建立一個HandlerContext並新增到列表

3. 新增context

4. 回撥新增事件

首先我們看第一步, 重複新增驗證

我們跟到checkMultiplicity(handler)中:

private static void checkMultiplicity(ChannelHandler handler) {
    if (handler instanceof ChannelHandlerAdapter) {
        ChannelHandlerAdapter h = (ChannelHandlerAdapter) handler; 
        if (!h.isSharable() && h.added) {
            throw new ChannelPipelineException(
                    h.getClass().getName() +
                    " is not a @Sharable handler, so can't be added or removed multiple times.");
        }
        //滿足條件設定為true, 代表已新增
        h.added = true;
    }
}

首先判斷是不是ChannelHandlerAdapter型別, 因為我們自定義的handler通常會直接或者間接的繼承該介面, 所以這裡為true

拿到handler之後轉換成ChannelHandlerAdapter型別, 然後進行條件判斷

 if (!h.isSharable() && h.added) 代表如果不是共享的handler, 並且是未新增狀態, 則丟擲異常:

我們可以跟到isSharable()方法中去:

public boolean isSharable() { 
    Class<?> clazz = getClass();
    Map<Class<?>, Boolean> cache = InternalThreadLocalMap.get().handlerSharableCache();
    Boolean sharable = cache.get(clazz);
    if (sharable == null) { 
        //如果這個類註解了Sharable.class, 說明這個類會被多個channel共享
        sharable = clazz.isAnnotationPresent(Sharable.class);
        cache.put(clazz, sharable);
    }
    return sharable;
}

首先拿到當前handler的class物件

然後再從netty自定義的一個ThreadLocalMap物件中獲取一個盛放handler的class物件的map, 並獲取其value

如果value值為空, 則會判斷是否被Sharable註解, 並將自身handler的class物件和判斷結果存入map物件中, 最後返回判斷結果

這說明了被Sharable註解的handler是一個共享handler

從這個邏輯我們可以判斷, 共享物件是可以重複新增的

我們回到checkMultiplicity(handler)方法中:

如果是共享物件或者沒有被新增, 則將ChannelHandlerAdapter的added設定為true, 代表已新增

剖析完了重複新增驗證, 回到addLast方法中, 我們看第二步, 建立一個HandlerContext並新增到列表:

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

首先看filterName(name, handler)方法, 這個方法是判斷新增handler的name是否重複

跟到filterName方法中:

private String filterName(String name, ChannelHandler handler) {
    if (name == null) {
        //沒有名字建立預設名字
        return generateName(handler);
    }
    //檢查名字是否重複
    checkDuplicateName(name);
    return name;
}

因為我們新增handler時候, 不一定會會給handler命名, 所以這一步name有可能是null, 如果是null, 則建立一個預設的名字, 這裡建立名字的方法我們就不往裡跟了, 有興趣的同學可以自己跟進去看

然後再檢查名字是否重複

我們跟到checkDuplicateName(name)這個方法中:

private void checkDuplicateName(String name) {
    //不為空
    if (context0(name) != null) {
        throw new IllegalArgumentException("Duplicate handler name: " + name);
    }
}

這裡有個context0(name)方法, 我們跟進去:

private AbstractChannelHandlerContext context0(String name) {
    //遍歷pipeline
    AbstractChannelHandlerContext context = head.next;
    while (context != tail) {
        //發現name相同, 說明存在handler
        if (context.name().equals(name)) {
            //返回
            return context;
        }
        context = context.next;
    }
    return null;
}

這裡做的操作非常簡單, 就是將pipeline中, 從head節點往下遍歷HandlerContext, 一直遍歷到tail, 如果發現名字相同則會認為重複並返回HandlerContext物件

我們回到addLast()方法中並繼續看新增建立相關的邏輯:

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

filterName(name, handler)這步如果並沒有重複則會返回handler的name

我們繼續跟到newContext(group, filterName(name, handler), handler)方法中:

private AbstractChannelHandlerContext newContext(EventExecutorGroup group, String name, ChannelHandler handler) {
    return new DefaultChannelHandlerContext(this, childExecutor(group), name, handler);
}

這裡我們看到建立了一個DefaultChannelHandlerContext物件, 構造方法的引數中, 第一個this代表當前的pipeline物件, group為null, 所以childExecutor(group)也會返回null, name為handler的名字, handler為新新增的handler物件

我們繼續跟到DefaultChannelHandlerContext的構造方法中:

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

我們看到首先呼叫了父類的構造方法, 之後將handler賦值為自身handler的成員變數, HandlerConext和handler關係在此也展現了出來, 是一種組合關係

我們首先看父類的構造方法, 有這麼兩個引數:isInbound(handler), isOutbound(handler), 這兩個引數意思是判斷需要新增的handler是inboundHandler還是outBoundHandler

跟到isInbound(handler)方法中:

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

這裡通過是否實現ChannelInboundHandler介面來判斷是否為inboundhandler

同樣我們看isOutbound(handler)方法:

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

通過判斷是否實現ChannelOutboundHandler介面判斷是否為outboundhandler

在跟到其父類AbstractChannelHandlerContext的構造方法中:

AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name, 
                              boolean inbound, boolean outbound) {
    this.name = ObjectUtil.checkNotNull(name, "name");
    this.pipeline = pipeline;
    this.executor = executor;
    this.inbound = inbound;
    this.outbound = outbound;
    ordered = executor == null || executor instanceof OrderedEventExecutor;
}

一切都不陌生了, 因為我們tail節點和head節點建立的時候同樣走到了這裡

這裡初始化了name, pipeline, 以及標識新增的handler是inboundhanlder還是outboundhandler

我們回到最初的addLast()方法中:

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        //判斷handler是否被重複新增(1)
        checkMultiplicity(handler);
        //建立一個HandlerContext並新增到列表(2)
        newCtx = newContext(group, filterName(name, handler), handler);

        //新增HandlerContext(3)
        addLast0(newCtx);

        //是否已註冊
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

        EventExecutor executor = newCtx.executor();
        if (!executor.inEventLoop()) {
            newCtx.setAddPending();
            //回撥使用者事件
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerAdded0(newCtx);
                }
            });
            return this;
        }
    }
    //回撥新增事件(4)
    callHandlerAdded0(newCtx);
    return this;
}

我們跟完了建立HandlerContext的相關邏輯, 我們繼續跟第三步, 新增HandlerContext

我們跟進addLast0(newCtx)中:

private void addLast0(AbstractChannelHandlerContext newCtx) {
    //拿到tail節點的前置節點
    AbstractChannelHandlerContext prev = tail.prev;
    //當前節點的前置節點賦值為tail節點的前置節點
    newCtx.prev = prev;
    //當前節點的下一個節點賦值為tail節點
    newCtx.next = tail;
    //tail前置節點的下一個節點賦值為當前節點
    prev.next = newCtx;
    //tail節點的前一個節點賦值為當前節點
    tail.prev = newCtx;
}

這一部分也非常簡單, 做了一個指標的指向操作, 將新新增的handlerConext放在tail節點之前, 之前tail節點的上一個節點之後, 熟悉雙向連結串列的同學對此邏輯應該不會陌生, 如果是第一次新增handler, 那麼新增後的結構入下圖所示:

4-2-1

新增完handler之後, 這裡會判斷當前channel是否已經註冊, 這部分邏輯我們之後再進行剖析, 我們繼續往下走

之後會判斷當前執行緒執行緒是否為eventLoop執行緒, 如果不是eventLoop執行緒, 就將添加回調事件封裝成task交給eventLoop執行緒執行, 否則, 直接執行添加回調事件callHandlerAdded0(newCtx)

跟進callHandlerAdded0(newCtx):

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    try {
        ctx.handler().handlerAdded(ctx);
        ctx.setAddComplete();
    } catch (Throwable t) {
        //忽略程式碼
    }
}

我們重點關注這句

ctx.handler().handlerAdded(ctx);

其中ctx是我們新建立的HandlerContext, 通過handler()方法拿到繫結的handler, 也就是新新增的handler, 然後執行handlerAdded(ctx)方法, 如果我們沒有重寫這個方法, 則會執行父類的該方法

在ChannelHandlerAdapter類中定義了該方法的實現:

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

我們看到沒做任何操作, 也就是如果我們沒有重寫該方法時, 如果新增handler之後將不會做任何操作, 這裡如果我們需要做一些業務邏輯, 可以通過重寫該方法進行實現

以上就是新增handler的有關的業務邏輯