国产xxxx99真实实拍_久久不雅视频_高清韩国a级特黄毛片_嗯老师别我我受不了了小说

資訊專欄INFORMATION COLUMN

【源起Netty 正傳】Netty Channel

jindong / 2817人閱讀

摘要:搞懂了這部分后,我們將明白在世界中扮演的角色進擊的此圖展示的已經算是優化后的了用到了線程池。多線程將這種處理操作分隔出來,非型操作業務操作配備以線程池,進化成多線程模型這樣的架構,系統瓶頸轉移至部分。

Channel定位

注意:如無特別說明,文中的Channel都指的是Netty Channel(io.netty.channel)

一周時間的Channel家族學習,一度讓我懷疑人生——研究這個方法有沒有用?學習Netty是不是有點兒下了高速走鄉間小路的意思?我為啥要讀源碼?
之所以產生這些疑問,除了我本身心理活動豐富以外,主要病因在于沒搞清楚Channel在Netty體系中的定位。而沒能清晰理解Netty的定位,也默默的送出了一記助攻。

作些本質思考:Netty是一個NIO框架,是一個嫁接在java NIO基礎上的框架

宏觀上可以這么理解,見下圖:

先不急著聊Channel,回顧下IO演進過程,重點關注IO框架的結構變化。搞懂了這部分后,我們將明白Channel在IO世界中扮演的角色!

進擊的IO BIO

此圖展示的已經算是優化后的BIO了——用到了線程池。顯然,每一個client都需要server端付出一個Thread的代價,即使你通過線程池做了優化,由于受到線程個數的制約,激增的客戶端依舊表現的“欲求不滿”。

NIO

Acceptor注冊Selector,監聽accept事件

當客戶端連接后,觸發accept事件

服務器構建對應的Channel,并在其上注冊Selector,監聽讀寫事件

當發生讀寫事件后,進行相應的讀寫處理

Reactor單線程

與NIO模型相似,當然也就有和NIO同樣的問題:selector/reactor單個線程處理多個channel的各種操作,如果其中一個channel的事件處理延緩了,將影響其它channel。

Reactor多線程

將read/write這種io處理操作分隔出來,非io型操作(業務操作)配備以線程池,進化成reactor多線程模型:

這樣的架構,系統瓶頸轉移至Reactor部分。而目前勞苦功高的Reactor作了兩件事:
1.接收客戶端鏈接請求
2.處理IO型讀寫操作

主從Reactor

將接收client鏈接的功能再次拆分出來:

Netty恰恰就是主從Reactor模型的實踐者,想想服務端創建時的代碼:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)

...

從nio時代的模型圖上開始出現channel(java channel),它的定位就是進行諸如connect、write、read、close等底層交互。概括一下,java channel是上承selector下連socket的存在。而netty channel,則把java channel當作了底層

源碼分析 類結構

清楚了Channel的定位,接下來對其常用api進行分析。

首先拍出類圖:

其實Channel內部還有一套體系,Unsafe家族:

Unsafe是Channel的內置類(接口),與java channel交互的重任最終會落到Unsafe身上。

write方法

write只是將數據寫入到了ChannelOutboundBuffer中,并沒有真正的發送出去,到flush方法調用時,才寫入到java channel中發送給對方。

下面列出AbstractChannel的write方法,值得關注的地方已打上中文注釋:

@Override
public final void write(Object msg, ChannelPromise promise) {
    assertEventLoop();

    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    if (outboundBuffer == null) {
        // If the outboundBuffer is null we know the channel was closed and so
        // need to fail the future right away. If it is not null the handling of the rest
        // will be done in flush0()
        // See https://github.com/netty/netty/issues/2362
        safeSetFailure(promise, WRITE_CLOSED_CHANNEL_EXCEPTION);
        // release message now to prevent resource-leak
        ReferenceCountUtil.release(msg);
        return;
    }

    int size;
    try {
        msg = filterOutboundMessage(msg);   //作消息的包裝,轉換成ByteBuf等
        size = pipeline.estimatorHandle().size(msg);
        if (size < 0) {
            size = 0;
        }
    } catch (Throwable t) {
        safeSetFailure(promise, t);
        ReferenceCountUtil.release(msg);
        return;
    }

    outboundBuffer.addMessage(msg, size, promise);    //msg消息寫入ChannelOutboundBuffer
}

上述代碼最后一行,msg寫入了ChannelOutboundBuffer的尾節點tailEntry,同時將unflushedEntry賦值暫存。代碼展開如下:

public void addMessage(Object msg, int size, ChannelPromise promise) {
    Entry entry = Entry.newInstance(msg, size, total(msg), promise);
    if (tailEntry == null) {
        flushedEntry = null;
        tailEntry = entry;
    } else {
        Entry tail = tailEntry;
        tail.next = entry;
        tailEntry = entry;
    }
    if (unflushedEntry == null) {    //注釋一、標記成“未刷新”的數據
        unflushedEntry = entry;
    }

    incrementPendingOutboundBytes(entry.pendingSize, false);
}
ChannelOutboundBuffer類

這里對ChannelOutboundBuffer類進行簡單說明,按慣例先看類注釋。

/**
 * (Transport implementors only) an internal data structure used by {@link AbstractChannel} to store its pending
 * outbound write requests.
 *
 *  省略...
 */

前文提到過,write方法將消息寫到ChannelOutboundBuffer,算是數據暫存;之后的flush再將消息刷到java channel乃至客戶端。

來張示意圖,方便理解:

圖中列出的三個屬性,在write->ChannelOutboundBuffer->flush的數據流轉過程中比較關鍵。Entry是啥?ChannelOutboundBuffer的靜態內部類,典型的鏈表結構數據:

static final class Entry {

        Entry next;
        
        // 省略...
}

write方法的最后部分(注釋一位置)調用outboundBuffer.addMessage(msg, size, promise),已將封裝msg的Entry賦值給tailEntry和unflushedEntry;而flush方法,通過調用outboundBuffer.addFlush()(下文,注釋二位置),將unflushedEntry間接賦值給了flushedEntry

public void addFlush() {
    Entry entry = unflushedEntry;
    if (entry != null) {
        if (flushedEntry == null) {
            // there is no flushedEntry yet, so start with the entry
            flushedEntry = entry;
        }
        do {
            flushed ++;
            if (!entry.promise.setUncancellable()) {
                // Was cancelled so make sure we free up memory and notify about the freed bytes
                int pending = entry.cancel();
                decrementPendingOutboundBytes(pending, false, true);
            }
            entry = entry.next;
        } while (entry != null);

        // All flushed so reset unflushedEntry
        unflushedEntry = null;
    }
}
flush方法

直接從AbstractChannel的flush方法開始(若以Channel的flush為開端會經pipeline,將有很長調用鏈,省略):

public final void flush() {
    assertEventLoop();

    ChannelOutboundBuffer outboundBuffer = this.outboundBuffer;
    if (outboundBuffer == null) {
        return;
    }

    outboundBuffer.addFlush();    //注釋二、標記成“已刷新”數據
    flush0();    //數據處理
}

outboundBuffer.addFlush()方法已經分析過了,跟蹤調用鏈flush0->doWrite,我們看下AbstractNioByteChanneldoWrite方法:

@Override
protected void doWrite(ChannelOutboundBuffer in) throws Exception {
    int writeSpinCount = config().getWriteSpinCount();    //自旋計數,限制循環次數,默認16
    do {
        Object msg = in.current();    //flushedEntry的msg
        if (msg == null) {
            // Wrote all messages.
            clearOpWrite();
            // Directly return here so incompleteWrite(...) is not called.
            return;
        }
        writeSpinCount -= doWriteInternal(in, msg);
    } while (writeSpinCount > 0);

    incompleteWrite(writeSpinCount < 0);
}

writeSpinCount是個自旋計數,類似于自旋鎖的設定,防止當前IO線程由于網絡等原因無盡執行寫操作,而使得線程假死,造成資源浪費

觀察doWriteInternal方法,關鍵處依舊中文注釋伺候:

private int doWriteInternal(ChannelOutboundBuffer in, Object msg) throws Exception {
    if (msg instanceof ByteBuf) {
        ByteBuf buf = (ByteBuf) msg;
        if (!buf.isReadable()) {    //writerIndex - readerIndex >0 ? true: flase
            in.remove();
            return 0;
        }

        final int localFlushedAmount = doWriteBytes(buf);   //返回實際寫入到java channel的字節數
        if (localFlushedAmount > 0) {   //寫入成功
            in.progress(localFlushedAmount);
            /**
             * 1.已經全部寫完,執行in.remove()
             * 2.“寫半包”場景,直接返回1。
             *   外層方法的自旋變量writeSpinCount遞減成15,輪詢再次執行本方法
             */
            if (!buf.isReadable()) {
                in.remove();
            }
            return 1;
        }
    } else if (msg instanceof FileRegion) {
    
        //“文件型”消息處理邏輯省略..
        
    } else {
        // Should not reach here.
        throw new Error();
    }
    return WRITE_STATUS_SNDBUF_FULL;    //發送緩沖區滿,值=Integer.MAX_VALUE
}

回到doWrite方法,最后執行了incompleteWrite(writeSpinCount < 0)

protected final void incompleteWrite(boolean setOpWrite) {
    // Did not write completely.
    if (setOpWrite) {
        setOpWrite();
    } else {
        // Schedule flush again later so other tasks can be picked up in the meantime
        Runnable flushTask = this.flushTask;
        if (flushTask == null) {
            flushTask = this.flushTask = new Runnable() {
                @Override
                public void run() {
                    flush();
                }
            };
        }
        eventLoop().execute(flushTask);
    }
}

這里的設定挺有意思:

如果 setOpWrite = writeSpinCount < 0 = true,即 doWriteInternal方法返回值 = WRITE_STATUS_SNDBUF_FULL(發送緩沖區滿)時,設置寫操作位:

protected final void setOpWrite() {
    final SelectionKey key = selectionKey();
    // Check first if the key is still valid as it may be canceled as part of the deregistration
    // from the EventLoop
    // See https://github.com/netty/netty/issues/2104
    if (!key.isValid()) {
        return;
    }
    final int interestOps = key.interestOps();
    if ((interestOps & SelectionKey.OP_WRITE) == 0) {
        key.interestOps(interestOps | SelectionKey.OP_WRITE);
    }
}

其實就是設置SelectionKey的OP_WRITE操作位,在selector/reactor下次輪詢的時候,將再次執行寫操作

如果 setOpWrite = writeSpinCount < 0 = false,即 doWriteInternal方法返回值 = 1,16次寫半包仍舊沒將消息發送出去,則通過定時器再次執行flush:

public Channel flush() {
    pipeline.flush();
    return this;
}

結論:前者由于發送緩沖區滿,已無法寫入數據,于是繼希望于selector的下次輪詢;后者則可能只是因為自旋次數少,引起的數據發送不完全,直接將任務再次放入pipeline,而無需等待selector。
這無疑是種優化,細節之處,功力盡顯!

感謝

高性能Server---Reactor模型

文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。

轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/68810.html

相關文章

  • 源起Netty 正傳】升級版卡車——ByteBuf

    摘要:之所以稱它為卡車,只因編程思想中有段比喻我們可以把它想象成一個煤礦,通道是一個包含煤層數據的礦藏,而緩沖器則是派送到礦藏中的卡車。那么升級版卡車,自然指的就是。結構和功能之所以再次打造了升級版的緩沖器,顯然是不滿中的某些弊端。 卡車 卡車指的是java原生類ByteBuffer,這兄弟在NIO界大名鼎鼎,與Channel、Selector的鐵三角組合構筑了NIO的核心。之所以稱它為卡車...

    Jason_Geng 評論0 收藏0
  • 源起Netty 前傳】Linux網絡模型小記

    摘要:非阻塞模型這種也很好理解,由阻塞的死等系統響應進化成多次調用查看數據就緒狀態。復用模型,以及它的增強版就屬于該種模型。此時用戶進程阻塞在事件上,數據就緒系統予以通知。信號驅動模型應用進程建立信號處理程序時,是非阻塞的。 引言 之前的兩篇文章 FastThreadLocal怎么Fast?、ScheduledThreadPoolExecutor源碼解讀 搞的我心力交瘁,且讀源碼過程中深感功...

    Null 評論0 收藏0
  • 源起Netty 外傳】ServiceLoader詳解

    摘要:答曰摸索直譯為服務加載器,最終目的是獲取的實現類。代碼走起首先,要有一個接口形狀接口介紹然后,要有該接口的實現類。期具體實現依靠的內部類,感性趣的朋友可以自己看一下。總結重點在于可跨越包獲取,這一點筆者通過多模塊項目親測延時加載特性 前戲 netty源碼注釋有云: ... If a provider class has been installed in a jar file tha...

    MoAir 評論0 收藏0
  • 源起Netty 外傳】System.getPropert()詳解

    摘要:閱讀源碼時,發現很多,理所當然會想翻閱資料后,該技能,姿勢如下環境中的全部屬性全部屬性注意如果將本行代碼放在自定義屬性之后,會不會打出把自定義屬性也給獲取到可以結論會獲取目前環境中全部的屬性值,無論系統提供還是個人定義系統提供屬性代碼中定義 閱讀源碼時,發現很多System.getProperty(xxx),理所當然會想:whats fucking this? 翻閱資料后,Get該技能...

    lixiang 評論0 收藏0
  • 源起Netty 外傳】FastThreadLocal怎么Fast?

    摘要:實現原理淺談幫助理解的示意圖中有一屬性,類型是的靜態內部類。剛剛說過,是一個中的靜態內部類,則是的內部節點。這個會在線程中,作為其屬性初始是一個數組的索引,達成與類似的效果。的方法被調用時,會根據記錄的槽位信息進行大掃除。 概述 FastThreadLocal的類名本身就充滿了對ThreadLocal的挑釁,快男FastThreadLocal是怎么快的?源碼中類注釋坦白如下: /** ...

    gxyz 評論0 收藏0

發表評論

0條評論

最新活動
閱讀需要支付1元查看
<