2. 程式人生 > >死磕Netty原始碼之ChannelPipeline原始碼解析(一)

死磕Netty原始碼之ChannelPipeline原始碼解析(一)

阿新 • • 發佈：2018-12-25

前言

ChannelPipeline資料管道是ChannelHandler資料處理器的容器，負責ChannelHandler的管理和事件的攔截與排程

原始碼分析

初始化

在Channel初始化時會呼叫到AbstractChannel的構造方法，Pipeline的初始化動作是在AbstractChannel構造方法中完成的

protected AbstractChannel(Channel parent) {
    this.parent = parent;
    id = newId();
    unsafe = newUnsafe();
    pipeline = newChannelPipeline();
}

接下來我們繼續跟進newChannelPipeline()，探究一下DefaultChannelPipeline底層實現細節

protected DefaultChannelPipeline newChannelPipeline() {
    return new DefaultChannelPipeline(this);
}

protected DefaultChannelPipeline(Channel channel) {
    this.channel = ObjectUtil.checkNotNull(channel, "channel");

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

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

在newChannelPipeline()中呼叫了DefaultChannelPipeline的構造方法，DefaultChannelPipeline構造方法主要完成了三件事。首先將與之關聯的Channel儲存在屬性Channel中，然後例項化了兩個物件(一個是TailContext例項tail，一個是HeadContext例項head)，然後將head和tail相互指向構成了一個雙向連結串列。資料結構如下圖
ChannelPipeline資料結構


DefaultChannelPipeline中的每個節點是一個ChannelHandlerContext物件(一個ChannelPipeline中可以有多個ChannelHandler例項，而每一個ChannelHandler例項與ChannelPipeline之間的橋樑就是ChannelHandlerContext例項)

新增節點

我們在初始化bootstrap的時候應該對下面這段程式碼不陌生，它是將ChannelHandler新增到pipeline的雙向連結串列中去的

bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        socketChannel.pipeline().addLast(new NettyMessageDecoder(serializer));
        socketChannel.pipeline().addLast(new NettyMessageEncoder(serializer));
    }
})

下面我們來跟下addLast方法

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

public final ChannelPipeline addLast(EventExecutorGroup executor, ChannelHandler... handlers) {
    if (handlers == null) {
        throw new NullPointerException("handlers");
    }
    for (ChannelHandler h: handlers) {
        if (h == null) {
            break;
        }
        addLast(executor, null, h);
    }
    return this;
}

public final ChannelPipeline addLast(EventExecutorGroup group, String name, ChannelHandler handler) {
    final AbstractChannelHandlerContext newCtx;
    synchronized (this) {
        // 1.禁止非Sharable的handler重複新增到不同的pipeline中
        checkMultiplicity(handler);

        // 2.建立節點 => DefaultChannelHandlerContext
        newCtx = newContext(group, filterName(name, handler), handler);

        // 3.新增節點
        addLast0(newCtx);

        // 如果channel沒有與eventloop繫結
        // 則建立一個任務 這個任務會在channel被register的時候呼叫
        if (!registered) {
            newCtx.setAddPending();
            callHandlerCallbackLater(newCtx, true);
            return this;
        }

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

這裡簡單地用synchronized方法是為了防止多執行緒併發操作pipeline底層的雙向連結串列

檢查是否有重複Handler

檢查是否有重複Handler的實現在checkMultiplicity()方法中

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.");
        }
        h.added = true;
    }
}

ChannelHandlerAdapter使用一個成員變數added標識一個Channel是否已經新增過，如果當前要新增的Handler是非共享並且已經新增過那就丟擲異常，否則標識該Handler已經新增。由此可見一個Handler如果是Sharable的就可以無限次被新增到Pipeline中，我們客戶端程式碼如果要讓一個Handler被共用，只需要加一個@Sharable標註即。而如果Handler是Sharable的一般就通過Spring的注入的方式使用，不需要每次都新建物件

建立節點DefaultChannelHandlerContext

建立節點呼叫了newContext()方法，程式碼如下

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

這裡我們需要先分析filterName(name, handler)這段程式碼，這個方法用於給handler建立一個唯一性的名字

private String filterName(String name, ChannelHandler handler) {
    if (name == null) {
        // 1.如果傳入的name為空 則生成
        // Netty生成的name預設為=> 簡單類名#0
        // 如果簡單類名#0已存在則將基數+1 生成name為簡單類名#1 以此遞增
        return generateName(handler);
    }
    // 2.檢查是否有重名 檢查通過則返回
    checkDuplicateName(name);
    return name;
}

檢查是否存在重名的Context，如果已存在重名的Context物件則丟擲異常

private void checkDuplicateName(String name) {
    // 判斷是否已有重名的Context
    if (context0(name) != null) {
        // 如果已存在重名的Context則丟擲異常
        throw new IllegalArgumentException("Duplicate handler name: " + name);
    }
}

private AbstractChannelHandlerContext context0(String name) {
    // 從頭結點開始遍歷所有的Context
    // 只要發現某個Context的名字與待新增的name相同 就返回該Context
    AbstractChannelHandlerContext context = head.next;
    while (context != tail) {
        if (context.name().equals(name)) {
            return context;
        }
        context = context.next;
    }
    return null;
}

在處理完name之後就進入到建立context的過程，程式碼如下

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

private EventExecutor childExecutor(EventExecutorGroup group) {
    if (group == null) {
        return null;
    }
    ...
}

由前面的呼叫鏈得知group為null，因此此處的childExecutor(group)也返回null

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

在DefaultChannelHandlerContext構造方法中標記了Handler是否是Inboud型別或OutBound型別並呼叫了父類的構造方法，最後儲存了Handler的引用，判斷Handler是Inbound還是OutBound的程式碼如下

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

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

呼叫父類建構函式，初始化成員變數

AbstractChannelHandlerContext(DefaultChannelPipeline pipeline, EventExecutor executor, String name, boolean inbound, boolean outbound) {
    this.name = ObjectUtil.checkNotNull(name, "name");
    this.pipeline = pipeline;
    // 此處exeutor為null
    this.executor = executor;
    this.inbound = inbound;
    this.outbound = outbound;
    // 初次ordered為true
    ordered = executor == null || executor instanceof OrderedEventExecutor;
}

如果一個Handler實現了兩類介面，那麼它既是一個inBound型別的Handler，又是一個outBound型別的Handler，如下圖
ChannelDuplexHandler類圖
常用的將decode操作和encode操作合併到一起的codec，一般會繼承MessageToMessageCodec，而MessageToMessageCodec就是繼承ChannelDuplexHandler

新增Context節點到Pipeline的雙向連結串列中

在建立完Context之後呼叫addLast0()將Context新增至Pipeline的雙向連結串列中

private void addLast0(AbstractChannelHandlerContext newCtx) {
    AbstractChannelHandlerContext prev = tail.prev;
    newCtx.prev = prev;
    newCtx.next = tail;
    prev.next = newCtx;
    tail.prev = newCtx;
}

新增節點其實就是個雙向連結串列的插入操作，具體步驟如下圖所示
Pipeline新增節點過程

操作完畢，該Context就加入到Pipeline中
Pipeline新增節點完成

至此Pipeline新增節點的操作就完成了，其他的addXXX方法原理類似

Context節點新增完畢回撥Handler方法

回撥使用者方法邏輯在callHandlerAdded0()方法中完成，程式碼如下

private void callHandlerAdded0(final AbstractChannelHandlerContext ctx) {
    try {
        ctx.handler().handlerAdded(ctx);
        ctx.setAddComplete();
    } catch (Throwable t) {
        ...
    }
}

Handler可以重寫handlerAdded方法在Handler被新增成功後 執行部分操作

public class DemoHandler extends SimpleChannelInboundHandler<...> {
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // 節點被新增完畢之後回撥到此
    }
}

接下來，設定該節點的狀態

final void setAddComplete() {
    for (;;) {
        int oldState = handlerState;
        if (oldState == REMOVE_COMPLETE || HANDLER_STATE_UPDATER.compareAndSet(this, oldState, ADD_COMPLETE)) {
            return;
        }
    }
}

用cas修改節點的狀態至REMOVE_COMPLETE(說明該節點已經被移除)或者ADD_COMPLETE(說明節點已被新增)

刪除節點

Netty有個最大的特性之一就是Handler可插拔做到動態編織Pipeline，比如在首次建立連線的時候需要通過進行許可權認證，在認證通過之後就可以將此Context移除，下次Pipeline在傳播事件的時候就就不會呼叫到許可權認證處理器

下面是許可權認證Handler最簡單的實現，第一個資料包傳來的是認證資訊，如果校驗通過就刪除此Handler，否則直接關閉連線

public class AuthHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf data) throws Exception {
        if (verify(authDataPacket)) {
            ctx.pipeline().remove(this);
        } else {
            ctx.close();
        }
    }

    private boolean verify(ByteBuf byteBuf) {
        // ...
    }
}

重點就在ctx.pipeline().remove(this)這段程式碼

public final ChannelPipeline remove(ChannelHandler handler) {
    remove(getContextOrDie(handler));
    return this;
}

remove操作可以分為如下三個步驟

1.找到待刪除的節點
2.調整雙向連結串列指標刪除節點
3.Context節點刪除完畢回撥Handler方法

找到待刪除的節點

private AbstractChannelHandlerContext getContextOrDie(String name) {
    AbstractChannelHandlerContext ctx = (AbstractChannelHandlerContext) context(name);
    if (ctx == null) {
        throw new NoSuchElementException(name);
    } else {
        return ctx;
    }
}

public final ChannelHandlerContext context(ChannelHandler handler) {
    if (handler == null) {
        throw new NullPointerException("handler");
    }

    AbstractChannelHandlerContext ctx = head.next;
    for (;;) {

        if (ctx == null) {
            return null;
        }

        if (ctx.handler() == handler) {
            return ctx;
        }

        ctx = ctx.next;
    }
}

通過遍歷連結串列方式根據Handler找到對應的Context節點(判斷依據 => Context的Handler和當前Handler相同)

調整雙向連結串列指標刪除節點

private AbstractChannelHandlerContext remove(final AbstractChannelHandlerContext ctx) {
    // 頭結點和為節點不能刪除
    assert ctx != head && ctx != tail;

    synchronized (this) {
        // 調整雙向連結串列指標刪除節點
        remove0(ctx);

        // 如果channel沒有與eventloop繫結
        // 則建立一個任務 這個任務會在channel被register的時候呼叫
        if (!registered) {
            callHandlerCallbackLater(ctx, false);
            return ctx;
        }

        // 回撥使用者函式
        EventExecutor executor = ctx.executor();
        if (!executor.inEventLoop()) {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    callHandlerRemoved0(ctx);
                }
            });
            return ctx;
        }
    }
    callHandlerRemoved0(ctx);
    return ctx;
}

刪除節點是通過remove0方法實現的，程式碼如下

private static void remove0(AbstractChannelHandlerContext ctx) {
    AbstractChannelHandlerContext prev = ctx.prev;
    AbstractChannelHandlerContext next = ctx.next;
    prev.next = next;
    next.prev = prev;
}

刪除節點其實就是個雙向連結串列的刪除操作，具體步驟如下圖所示
Pipeline刪除節點過程

操作完畢，該Context就從Pipeline中移除
Pipeline刪除節點完成

Context節點刪除完畢回撥Handler方法

回撥使用者方法邏輯在callHandlerRemoved0()方法中完成，程式碼如下

private void callHandlerRemoved0(final AbstractChannelHandlerContext ctx) {
    try {
        try {
            ctx.handler().handlerRemoved(ctx);
        } finally {
            ctx.setRemoved();
        }
    } catch (Throwable t) {
        fireExceptionCaught(new ChannelPipelineException(ctx.handler().getClass().getName() + ".handlerRemoved() has thrown an exception.", t));
    }
}

Handler可以重寫handlerRemoved方法在Handler被刪除成功後 執行部分操作

public class DemoHandler extends SimpleChannelInboundHandler<...> {
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        // 節點被刪除完畢之後回撥到此
    }
}

最後將該節點的狀態設定為removed

final void setRemoved() {
    handlerState = REMOVE_COMPLETE;
}

至此Pipeline新增節點的刪除就完成了，其他的removeXXX方法原理類似

相關推薦

死磕Netty原始碼之ChannelPipeline原始碼解析(二)

前言 在上一篇部落格中我們介紹了Pipeline資料結構以及節點的相關操作，在這一篇部落格中將介紹Pipeline的事件傳播 事件傳播 inBound事件傳播 我們回顧一下Netty新連線接入的過程，在三次握手成功後新連線會呼叫pipelin

死磕Netty原始碼之ChannelPipeline原始碼解析(一)

前言 ChannelPipeline資料管道是ChannelHandler資料處理器的容器，負責ChannelHandler的管理和事件的攔截與排程 原始碼分析 初始化 在Channel初始化時會呼叫到AbstractChannel的構造方法

死磕 java集合之DelayQueue原始碼分析

問題 （1）DelayQueue是阻塞佇列嗎？ （2）DelayQueue的實現方式？ （3）DelayQueue主要用於

死磕Netty原始碼之服務端啟動原始碼解析

前言 本部落格講述的是Netty是如何繫結埠、啟動服務。啟動服務的過程中你將會了解到Netty各大核心元件 服務端啟動DEMO 先從一個簡單的服務端啟動DEMO開始，以下是一個標準的Netyy服務端程式碼 public final class

死磕Netty原始碼之Netty編解碼原始碼解析

前言 Netty的底層是基於TCP實現的，TCP協議在傳輸資料的過程中一個完整的業務可能會被TCP拆分成多個包進行傳送，也有可能把多個小的包封裝成一個大的資料包傳送，因此我們需要考慮Netty的粘包拆包問題 訊息解碼器 Netty提供了拆包的基類Byt

死磕Netty原始碼之新連線接入原始碼解析

前言 本部落格主要是介紹Netty在新連線接入後的相關處理 新連線建立 新連線建立可以分為以下三個步驟 1.檢測到有新的連線 2.將新的連線註冊到Worker執行緒組 3.註冊新連線的讀事件 在Reactor執行緒模型詳解部落格中我們已經知

死磕Netty原始碼之記憶體分配詳解(四)PoolArena全域性記憶體分配

記憶體分配 全域性分配 記憶體池的初始階段執行緒是沒有記憶體快取的，所以最開始的記憶體分配都需要在全域性分配區進行分配 全域性分配區的記憶體構造和執行緒私有分配區的類似(包含Tiny、Small、Normal幾種規模 計算索引的方式也都是一模一樣的

死磕Netty原始碼之記憶體分配詳解(三)PoolThreadCache執行緒快取記憶體分配

記憶體分配 執行緒私有分配 在介紹PoolArena記憶體分配結構分析的時候提到記憶體分配會先從執行緒快取裡分配，這個執行緒快取其實就是PoolThreadCache PoolThreadCache 成員變數 final PoolA

死磕Netty原始碼之記憶體分配詳解(二)PoolArena記憶體分配結構分析

前言 在應用層通過設定PooledByteBufAllocator來執行ByteBuf的分配，但是最終的記憶體分配工作被委託給PoolArena。由於Netty通常用於高併發系統所以各個執行緒進行記憶體分配時競爭不可避免，這可能會極大的影響記憶體分配的效率，為

死磕Netty原始碼之記憶體分配詳解(一)(PooledByteBufAllocator)

前言 為了避免頻繁的記憶體分配給系統帶來負擔以及GC對系統性能帶來波動，Netty4使用了記憶體池來管理記憶體的分配和回收，Netty記憶體池參考了Slab分配和Buddy分配思想。Slab分配是將記憶體分割成大小不等的記憶體塊，在使用者執行緒請求時根據請求的

死磕Netty原始碼之Reactor執行緒模型詳解(二)NioEventLoop的執行

前言 在上一篇部落格介紹了NioEventLoop的啟動流程，這篇部落格將介紹NioEventLoop的執行流程 NioEventLoop執行 NioEventLoop的執行在run方法中完成，程式碼如下 protected void run()

死磕Netty原始碼之Reactor執行緒模型詳解(一)NioEventLoop的啟動

前言 Netty中最核心的就是Reactor執行緒，對應Netty中的程式碼就是NioEventLoop。NioEventLoop是通過NioEventLoopGroup進行維護的，所以在介紹NioEventLoop前我們先介紹一下NioEventLoopGr

jdk1.7原始碼之-hashMap原始碼解析

背景： 筆者最近這幾天在思考，為什麼要學習設計模式，學些設計模式無非是提高自己的開發技能，但是通過這一段時間來看，其實我也學習了一些設計模式，但是都是一些demo，沒有具體的例子，學習起來不深刻，所以我感覺我可能要換一條路走，所以我現在想法是看一些原始碼的東西，一方面是因為自己大部分的原始碼其實沒有看過，另

Netty學習之旅------原始碼分析Netty執行緒本地分配機制與PooledByteBuf執行緒級物件池原理分析

final PoolArena<byte[]> heapArena; //使用輪叫輪詢機制，每個執行緒從heapArena[]中獲取一個，用於記憶體分配。 final PoolArena<ByteBuffer> directArena;

Netty學習之旅----原始碼分析Netty記憶體洩漏檢測

1、圖說Netty直接記憶體管理 2、Netty 直接記憶體的使用示例 ByteBuf buf = Unpooled.directBuffer(512); System.out.println(buf); // Si

Netty學習之旅------原始碼分析Netty解碼編碼器實現原理

package io.netty.handler.codec; import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandler; import io.netty.c

vue原始碼之入口檔案解析

由於專案中經常使用vue，所以這次趁有機會趕緊拜讀下原始碼，體驗下vue原始碼的設計風采。 一、下載原始碼 二、原始碼專案目錄 三、載入core核心入口檔案index.js 通過看原始碼我們可以得知這個入口檔案： （1）引入了幾個物件，包括Vue

Netty學習之旅----原始碼分析記憶體分配與釋放原理

static PooledHeapByteBuf newInstance(int maxCapacity) { PooledHeapByteBuf buf = RECYCLER.get(); buf.setRefCnt(1); buf.maxCapacity(m

Netty中的ChannelPipeline原始碼分析

ChannelPipeline在Netty中是用來處理請求的責任鏈，預設實現是DefaultChannelPipeline，其構造方法如下： 1 private final Channel channel; 2 private final ChannelFuture succeededFuture;

【go原始碼分析】go原始碼之slice原始碼分析

Go 語言切片是對陣列的抽象。 Go 陣列的長度不可改變，與陣列相比切片的長度是不固定的，可以追加元素，在追加時可能使切片的容量增大。 len() 和 cap() 函式     切片是可索引的，並且可以由 len() 方法獲取長度。