Netty源碼分析第4章(pipeline)---->第7節: 前章節內容回顧

 

Netty源碼分析第四章: pipeline

 

第七節: 前章節內容回顧

 

咱們在第一章和第三章中, 遺留了不少有關事件傳輸的相關邏輯, 這裏帶你們一一回顧

首先看兩個問題:

1.在客戶端接入的時候, NioMessageUnsafe的read方法中pipeline.fireChannelRead(readBuf.get(i))爲何會調用到ServerBootstrap的內部類ServerBootstrapAcceptor中的channelRead()方法

2.客戶端handler是何時被添加的?

首先看第一個問題:

1.在客戶端接入的時候, NioMessageUnsafe的read方法中pipeline.fireChannelRead(readBuf.get(i))爲何會調用到ServerBootstrap的內部類ServerBootstrapAcceptor中的channelRead()方法?

咱們首先看這段代碼:

public void read() { //必須是NioEventLoop方法調用的, 不能經過外部線程調用
    assert eventLoop().inEventLoop(); //服務端channel的config
    final ChannelConfig config = config(); //服務端channel的pipeline
    final ChannelPipeline pipeline = pipeline(); //處理服務端接入的速率
    final RecvByteBufAllocator.Handle allocHandle = unsafe().recvBufAllocHandle(); //設置配置
 allocHandle.reset(config); boolean closed = false; Throwable exception = null; try { try { do { //建立jdk底層的channel //readBuf用於臨時承載讀到連接
                int localRead = doReadMessages(readBuf); if (localRead == 0) { break; } if (localRead < 0) { closed = true; break; } //分配器將讀到的連接進行計數
 allocHandle.incMessagesRead(localRead); //鏈接數是否超過最大值
            } while (allocHandle.continueReading()); } catch (Throwable t) { exception = t; } int size = readBuf.size(); //遍歷每一條客戶端鏈接
        for (int i = 0; i < size; i ++) { readPending = false; //傳遞事件, 將建立NioSokectChannel進行傳遞 //最終會調用ServerBootstrap的內部類ServerBootstrapAcceptor的channelRead()方法  pipeline.fireChannelRead(readBuf.get(i)); } readBuf.clear(); allocHandle.readComplete(); pipeline.fireChannelReadComplete(); //代碼省略
    } finally { //代碼省略
 } }

重點看pipeline.fireChannelRead(readBuf.get(i))

首先, 這裏pipeline是服務端channel的pipeline, 也就是NioServerSocketChannel的pipeline

咱們學習過pipeline以後, 對這種寫法並不陌生, 就是傳遞channelRead事件, 這裏經過傳遞channelRead事件走到了ServerBootstrapAcceptor的channelRead()方法, 說明在這步以前, ServerBootstrapAcceptor做爲一個handler添加到了服務端channel的pipeline中, 那麼這個handler何時添加的呢?

咱們回顧下第一章, 初始化NioServerSocketChannel的時候, 調用了ServerBootstrap的init方法:

void init(Channel channel) throws Exception { //獲取用戶定義的選項(1)
    final Map<ChannelOption<?>, Object> options = options0(); synchronized (options) { channel.config().setOptions(options); } //獲取用戶定義的屬性(2)
    final Map<AttributeKey<?>, Object> attrs = attrs0(); synchronized (attrs) { for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) { @SuppressWarnings("unchecked") AttributeKey<Object> key = (AttributeKey<Object>) e.getKey(); channel.attr(key).set(e.getValue()); } } //獲取channel的pipline(3)
    ChannelPipeline p = channel.pipeline(); //work線程組(4)
    final EventLoopGroup currentChildGroup = childGroup; //用戶設置的Handler(5)
    final ChannelHandler currentChildHandler = childHandler; final Entry<ChannelOption<?>, Object>[] currentChildOptions; final Entry<AttributeKey<?>, Object>[] currentChildAttrs; //選項轉化爲Entry對象(6)
    synchronized (childOptions) { currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size())); } //屬性轉化爲Entry對象(7)
    synchronized (childAttrs) { currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size())); } //添加服務端handler(8)
    p.addLast(new ChannelInitializer<Channel>() { //初始化channel
 @Override public void initChannel(Channel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { @Override public void run() { pipeline.addLast(new ServerBootstrapAcceptor( currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); }

這個方法比較長, 咱們重點關注第8步, 添加服務端channel, 這裏的pipeline, 是服務服務端channel的pipeline, 也就是NioServerSocketChannel綁定的pipeline, 這裏添加了一個ChannelInitializer類型的handler

咱們看一下ChannelInitializer這個類的繼承關係:

public abstract class ChannelInitializer<C extends Channel> extends ChannelInboundHandlerAdapter { //省略類體
}

咱們看到其繼承了ChannelInboundHandlerAdapter, 說明是一個inbound類型的handler

這裏咱們可能會想到, 添加完handler會執行handlerAdded, 而後再handlerAdded方法中作了添加ServerBootstrapAcceptor這個handler

可是, 實際上並非這樣的, 當程序執行到這裏, 並無立刻執行handlerAdded, 咱們緊跟addLast方法

最後會跟到DefualtChannelPipeline的一個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; }

首先完成了handler的添加, 可是並無立刻執行回調

這裏咱們重點關注if (!registered)這個條件判斷, 其實在註冊完成, registered會變成true, 可是走到這一步的時候NioServerSockeChannel並無完成註冊(能夠回顧第一章看註冊在哪一步), 因此會進到if裏並返回自身

咱們重點關注callHandlerCallbackLater這個方法, 咱們跟進去:

private void callHandlerCallbackLater(AbstractChannelHandlerContext ctx, boolean added) { assert !registered; //判斷是否已添加, 未添加, 進行添加, 已添加進行刪除
    PendingHandlerCallback task = added ? new PendingHandlerAddedTask(ctx) : new PendingHandlerRemovedTask(ctx); //獲取第一個Callback任務
    PendingHandlerCallback pending = pendingHandlerCallbackHead; //若是第一個Callback任務爲空
    if (pending == null) { //將第一個任務設置爲剛建立的任務
        pendingHandlerCallbackHead = task; } else { while (pending.next != null) { pending = pending.next; } pending.next = task; } }

因咱們調用這個方法的時候added傳的true, 因此PendingHandlerCallback task賦值爲new PendingHandlerAddedTask(ctx)

PendingHandlerAddedTask這個類, 咱們從名字能夠看出, 這是一個handler添加的延遲任務, 用於執行handler延遲添加的操做, 一樣也對應一個名字爲PendingHandlerRemovedTask的類, 用於執行延遲刪除handler的操做, 這兩個類都繼承抽象類PendingHandlerCallback

 

咱們看PendingHandlerAddedTask類構造方法:

PendingHandlerAddedTask(AbstractChannelHandlerContext ctx) { super(ctx); }

這裏調用了父類的構造方法, 再跟進去:

PendingHandlerCallback(AbstractChannelHandlerContext ctx) { this.ctx = ctx; }

在父類中, 保存了要添加的context, 也就是ChannelInitializer類型的包裝類

回到callHandlerCallbackLater方法中:

PendingHandlerCallback pending = pendingHandlerCallbackHead;

這表示獲取第一個PendingHandlerCallback的任務, 其實PendingHandlerCallback是一個單向鏈表, 自身維護一個PendingHandlerCallback類型的next, 指向下一個任務, 在DefaultChannelPipeline這個類中, 定義了個PendingHandlerCallback類型的引用pendingHandlerCallbackHead, 用來指向延遲迴調任務的中的第一個任務

 

以後判斷這個任務是爲空, 若是是第一次添加handler, 那麼這裏就是空, 因此將第一個任務賦值爲咱們剛建立的添加任務

若是不是第一次添加handler, 則將咱們新建立的任務添加到鏈表的尾部, 由於這裏咱們是第一次添加, 因此第一個回調任務就指向了咱們建立的添加handler的任務

完成這一系列操做以後, addLast方法返歸, 此時並無完成添加操做

而何時完成添加操做的呢?

在服務端channel註冊時候的會走到AbstractChannel的register0方法:

private void register0(ChannelPromise promise) { try { //作實際的註冊(1)
 doRegister(); neverRegistered = false; registered = true; //觸發事件(2)
 pipeline.invokeHandlerAddedIfNeeded(); safeSetSuccess(promise); //觸發註冊成功事件(3)
 pipeline.fireChannelRegistered(); if (isActive()) { if (firstRegistration) { //傳播active事件(4)
 pipeline.fireChannelActive(); } else if (config().isAutoRead()) { beginRead(); } } } catch (Throwable t) { //省略代碼
 } }

重點關注第二步pipeline.invokeHandlerAddedIfNeeded(), 這裏已經經過doRegister()方法完成了實際的註冊, 咱們跟到該方法中:

final void invokeHandlerAddedIfNeeded() { assert channel.eventLoop().inEventLoop(); if (firstRegistration) { firstRegistration = false; callHandlerAddedForAllHandlers(); } }

這裏會判斷是否第一次註冊, 這裏返回true, 而後會執行callHandlerAddedForAllHandlers()方法, 咱們跟進去:

private void callHandlerAddedForAllHandlers() { final PendingHandlerCallback pendingHandlerCallbackHead; synchronized (this) { assert !registered; registered = true; pendingHandlerCallbackHead = this.pendingHandlerCallbackHead; this.pendingHandlerCallbackHead = null; } //獲取task
    PendingHandlerCallback task = pendingHandlerCallbackHead; while (task != null) { //執行添加handler方法
 task.execute(); task = task.next; } }

這裏拿到第一個延遲執行handler添加的task其實就是咱們以前剖析過的, 延遲執行handler添加的task, 就是PendingHandlerAddedTask對象

在while循環中, 經過執行execute()方法將handler添加

咱們跟到PendingHandlerAddedTask的execute()方法中:

void execute() { //獲取當前eventLoop線程
    EventExecutor executor = ctx.executor(); //是當前執行的線程
    if (executor.inEventLoop()) { callHandlerAdded0(ctx); } else { try { //添加到隊列
            executor.execute(this); } catch (RejectedExecutionException e) { //代碼省略
 } } }

終於在這裏, 咱們看到了執行回調的方法

再回到init方法中:

void init(Channel channel) throws Exception { //獲取用戶定義的選項(1)
    final Map<ChannelOption<?>, Object> options = options0(); synchronized (options) { channel.config().setOptions(options); } //獲取用戶定義的屬性(2)
    final Map<AttributeKey<?>, Object> attrs = attrs0(); synchronized (attrs) { for (Entry<AttributeKey<?>, Object> e: attrs.entrySet()) { @SuppressWarnings("unchecked") AttributeKey<Object> key = (AttributeKey<Object>) e.getKey(); channel.attr(key).set(e.getValue()); } } //獲取channel的pipline(3)
    ChannelPipeline p = channel.pipeline(); //work線程組(4)
    final EventLoopGroup currentChildGroup = childGroup; //用戶設置的Handler(5)
    final ChannelHandler currentChildHandler = childHandler; final Entry<ChannelOption<?>, Object>[] currentChildOptions; final Entry<AttributeKey<?>, Object>[] currentChildAttrs; //選項轉化爲Entry對象(6)
    synchronized (childOptions) { currentChildOptions = childOptions.entrySet().toArray(newOptionArray(childOptions.size())); } //屬性轉化爲Entry對象(7)
    synchronized (childAttrs) { currentChildAttrs = childAttrs.entrySet().toArray(newAttrArray(childAttrs.size())); } //添加服務端handler(8)
    p.addLast(new ChannelInitializer<Channel>() { //初始化channel
 @Override public void initChannel(Channel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { @Override public void run() { pipeline.addLast(new ServerBootstrapAcceptor( currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); } }); }

咱們繼續看第8步添加服務端handler

由於這裏的handler是ChannelInitializer, 因此完成添加以後會調用ChannelInitializer的handlerAdded方法

跟到handlerAdded方法:

public void handlerAdded(ChannelHandlerContext ctx) throws Exception { //默認狀況下, 會返回true
    if (ctx.channel().isRegistered()) { initChannel(ctx); } }

由於執行到這步服務端channel已經完成註冊, 因此會執行到initChannel方法

跟到initChannel方法:

private boolean initChannel(ChannelHandlerContext ctx) throws Exception { //這段代碼是否被執行過
    if (initMap.putIfAbsent(ctx, Boolean.TRUE) == null) { try { initChannel((C) ctx.channel()); } catch (Throwable cause) { exceptionCaught(ctx, cause); } finally { //調用以後會刪除當前節點
 remove(ctx); } return true; } return false; }

咱們關注initChannel這個方法, 這個方法是在ChannelInitializer的匿名內部來實現的, 這裏咱們注意, 在initChannel方法執行完畢以後會調用remove(ctx)刪除當前節點

咱們繼續跟進initChannel方法:

@Override public void initChannel(Channel ch) throws Exception { final ChannelPipeline pipeline = ch.pipeline(); ChannelHandler handler = config.handler(); if (handler != null) { pipeline.addLast(handler); } ch.eventLoop().execute(new Runnable() { @Override public void run() { pipeline.addLast(new ServerBootstrapAcceptor( currentChildGroup, currentChildHandler, currentChildOptions, currentChildAttrs)); } }); }

這裏首先添加用戶自定義的handler, 這裏若是用戶沒有定義, 則添加不成功, 而後, 會調用addLast將ServerBootstrapAcceptor這個handler添加了進去, 一樣這個handler也繼承了ChannelInboundHandlerAdapter, 在這個handler中, 重寫了channelRead方法, 因此, 這就是第一個問題的答案

緊接着咱們看第二個問題:

2.客戶端handler是何時被添加的?

咱們這裏看ServerBootstrapAcceptor的channelRead方法:

public void channelRead(ChannelHandlerContext ctx, Object msg) { final Channel child = (Channel) msg; //添加channelHadler, 這個channelHandler, 就是用戶代碼添加的ChannelInitializer
 child.pipeline().addLast(childHandler); //代碼省略

    try { //work線程註冊channel
        childGroup.register(child).addListener(new ChannelFutureListener() { //代碼省略
 }); } catch (Throwable t) { forceClose(child, t); } }

這裏真相能夠大白了, 服務端再建立完客戶端channel以後, 將新建立的NioSocketChannel做爲參數觸發channelRead事件(能夠回顧NioMessageUnsafe的read方法, 代碼這裏就不貼了), 因此這裏的參數msg就是NioSocketChannel

拿到channel時候再將客戶端的handler添加進去, 咱們回顧客戶端handler的添加過程:

.childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) { ch.pipeline().addLast(new StringDecoder()); ch.pipeline().addLast(new StringEncoder()); ch.pipeline().addLast(new ServerHandler()); } });

和服務端channel的邏輯同樣, 首先會添加ChannelInitializer這個handler可是沒有註冊因此沒有執行添加handler的回調, 將任務保存到一個延遲迴調的task中

等客戶端channel註冊完畢, 會將執行添加handler的回調, 也就是handlerAdded方法, 在回調中執行initChannel方法將客戶端handler添加進去, 而後刪除ChannelInitializer這個handler

由於在服務端channel中這塊邏輯已經進行了詳細的剖析, 因此這邊就不在贅述, 同窗們能夠本身跟進去走一遍流程

這裏注意, 由於每建立一個NioSoeketChannel都會調用服務端ServerBootstrapAcceptor的channelRead方法, 因此這裏會將每個NioSocketChannel的handler進行添加

 

第四章總結

        本章剖析了事件傳輸的相關邏輯, 包括handler的添加, 刪除, inbound和outbound以及異常事件的傳輸, 最後結合第一章和第三章, 剖析了服務端channel和客戶端channel的添加過程, 同窗們能夠課後跟進源碼, 將這些功能本身再走一遍以加深印象.其餘的有關事件傳輸的邏輯, 能夠結合這一章的知識點進行自行剖析

 

上一節: 傳播異常事件

下一節: AbstractByteBuf