深入淺出Netty：read

本系列：

boss執行緒主要負責監聽並處理accept事件，將socketChannel註冊到work執行緒的selector，由worker執行緒來監聽並處理read事件，本節主要分析Netty如何處理read事件。

accept->read

當work執行緒的selector檢測到OP_READ事件發生時，觸發read操作。

//NioEventLoop  
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {  
    unsafe.read();  
    if (!ch.isOpen()) {  
        // Connection already closed - no need to handle write.  
        return;  
    }  
}

12345678

//NioEventLoop if((readyOps&(SelectionKey.OP_READ|SelectionKey.OP_ACCEPT))!=0||readyOps==0){unsafe.read();if(!ch.isOpen()){// Connection already closed - no need to handle write. return;}}

該read方法定義在類NioByteUnsafe中。

//AbstractNioByteChannel.NioByteUnsafe
public final void read() {
    final ChannelConfig config = config();
    if (!config.isAutoRead() && !isReadPending()) {
        // ChannelConfig.setAutoRead(false) was called in the meantime
        removeReadOp();
        return;
    }

    final ChannelPipeline pipeline = pipeline();
    final ByteBufAllocator allocator = config.getAllocator();
    final int maxMessagesPerRead = config.getMaxMessagesPerRead();
    RecvByteBufAllocator.Handle allocHandle = this.allocHandle;
    if (allocHandle == null) {
        this.allocHandle = allocHandle = config.getRecvByteBufAllocator().newHandle();
    }

    ByteBuf byteBuf = null;
    int messages = 0;
    boolean close = false;
    try {
        int totalReadAmount = 0;
        boolean readPendingReset = false;
        do {
            byteBuf = allocHandle.allocate(allocator);
            int writable = byteBuf.writableBytes();
            int localReadAmount = doReadBytes(byteBuf);
            if (localReadAmount <= 0) {
                // not was read release the buffer
                byteBuf.release();
                byteBuf = null;
                close = localReadAmount < 0;
                break;
            }
            if (!readPendingReset) {
                readPendingReset = true;
                setReadPending(false);
            }
            pipeline.fireChannelRead(byteBuf);
            byteBuf = null;

            if (totalReadAmount >= Integer.MAX_VALUE - localReadAmount) {
                // Avoid overflow.
                totalReadAmount = Integer.MAX_VALUE;
                break;
            }

            totalReadAmount += localReadAmount;

            // stop reading
            if (!config.isAutoRead()) {
                break;
            }

            if (localReadAmount < writable) {
                // Read less than what the buffer can hold,
                // which might mean we drained the recv buffer completely.
                break;
            }
        } while (++ messages < maxMessagesPerRead);

        pipeline.fireChannelReadComplete();
        allocHandle.record(totalReadAmount);

        if (close) {
            closeOnRead(pipeline);
            close = false;
        }
    } catch (Throwable t) {
        handleReadException(pipeline, byteBuf, t, close);
    } finally {
        // Check if there is a readPending which was not processed yet.
        // This could be for two reasons:
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
        // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
        //
        // See https://github.com/netty/netty/issues/2254
        if (!config.isAutoRead() && !isReadPending()) {
            removeReadOp();
        }
    }
}

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182

//AbstractNioByteChannel.NioByteUnsafepublicfinalvoidread(){finalChannelConfig config=config();if(!config.isAutoRead

()&&!isReadPending()){// ChannelConfig.setAutoRead(false) was called in the meantimeremoveReadOp();return;}finalChannelPipeline pipeline=pipeline();finalByteBufAllocator allocator=config.getAllocator();finalintmaxMessagesPerRead=config.getMaxMessagesPerRead();RecvByteBufAllocator.Handle allocHandle=this.allocHandle;if(allocHandle==null){this.allocHandle=allocHandle=config.getRecvByteBufAllocator().newHandle();}ByteBuf byteBuf=null;intmessages=0;booleanclose=false;try{inttotalReadAmount=0;booleanreadPendingReset=false;do{byteBuf=allocHandle.allocate(allocator);intwritable=byteBuf.writableBytes();intlocalReadAmount=doReadBytes(byteBuf);if(localReadAmount<=0){// not was read release the bufferbyteBuf.release();byteBuf=null;close=localReadAmount<0;break;}if(!readPendingReset){readPendingReset=true;setReadPending(false);}pipeline.fireChannelRead(byteBuf);byteBuf=null;if(totalReadAmount>=Integer.MAX_VALUE-localReadAmount){// Avoid overflow.totalReadAmount=Integer.MAX_VALUE;break;}totalReadAmount+=localReadAmount;// stop readingif(!config.isAutoRead()){break;}if(localReadAmount<writable){// Read less than what the buffer can hold,// which might mean we drained the recv buffer completely.break;}}while(++messages<maxMessagesPerRead);pipeline.fireChannelReadComplete();allocHandle.record(totalReadAmount);if(close){closeOnRead(pipeline);close=false;}}catch(Throwablet){handleReadException(pipeline,byteBuf,t,close);}finally{// Check if there is a readPending which was not processed yet.// This could be for two reasons:// * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method// * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method//// See https://github.com/netty/netty/issues/2254if(!config.isAutoRead()&&!isReadPending()){removeReadOp();}}}

1、allocHandle負責自適應調整當前快取分配的大小，以防止快取分配過多或過少，先看看AdaptiveRecvByteBufAllocator內部實現：

public class AdaptiveRecvByteBufAllocator implements RecvByteBufAllocator {
    static final int DEFAULT_MINIMUM = 64;
    static final int DEFAULT_INITIAL = 1024;
    static final int DEFAULT_MAXIMUM = 65536;
    private static final int INDEX_INCREMENT = 4;
    private static final int INDEX_DECREMENT = 1;
    private static final int[] SIZE_TABLE;
}

12345678

publicclassAdaptiveRecvByteBufAllocatorimplementsRecvByteBufAllocator{staticfinalintDEFAULT_MINIMUM=64;staticfinalintDEFAULT_INITIAL=1024;staticfinalintDEFAULT_MAXIMUM=65536;privatestaticfinalintINDEX_INCREMENT=4;privatestaticfinalintINDEX_DECREMENT=1;privatestaticfinalint[]SIZE_TABLE;}

SIZE_TABLE：按照從小到大的順序預先儲存可以分配的快取大小。
從16開始，每次累加16，直到496，接著從512開始，每次增大一倍，直到溢位。
DEFAULT_MINIMUM：最小快取（64），在SIZE_TABLE中對應的下標為3。
DEFAULT_MAXIMUM ：最大快取（65536），在SIZE_TABLE中對應的下標為38。
DEFAULT_INITIAL ：初始化快取大小，第一次分配快取時，由於沒有上一次實際收到的位元組數做參考，需要給一個預設初始值。
INDEX_INCREMENT：上次預估快取偏小，下次index的遞增值。
INDEX_DECREMENT ：上次預估快取偏大，下次index的遞減值。

2、allocHandle.allocate(allocator) 申請一塊指定大小的記憶體。

//AdaptiveRecvByteBufAllocator.HandleImpl
public ByteBuf allocate(ByteBufAllocator alloc) {
    return alloc.ioBuffer(nextReceiveBufferSize);
}

1234	//AdaptiveRecvByteBufAllocator.HandleImplpublicByteBuf allocate(ByteBufAllocator alloc){returnalloc.ioBuffer(nextReceiveBufferSize);}

通過ByteBufAllocator的ioBuffer方法申請快取。

//AbstractByteBufAllocator
public ByteBuf ioBuffer(int initialCapacity) {
    if (PlatformDependent.hasUnsafe()) {
        return directBuffer(initialCapacity);
    }
    return heapBuffer(initialCapacity);
}

1234567

//AbstractByteBufAllocatorpublicByteBuf ioBuffer(intinitialCapacity){if(PlatformDependent.hasUnsafe()){returndirectBuffer(initialCapacity);}returnheapBuffer(initialCapacity);}

根據平臺是否支援unsafe，選擇使用直接實體記憶體還是堆上記憶體。

direct buffer方案：

//AbstractByteBufAllocator
public ByteBuf directBuffer(int initialCapacity) {
    return directBuffer(initialCapacity, Integer.MAX_VALUE);
}

public ByteBuf directBuffer(int initialCapacity, int maxCapacity) {
    if (initialCapacity == 0 && maxCapacity == 0) {
        return emptyBuf;
    }
    validate(initialCapacity, maxCapacity);
    return newDirectBuffer(initialCapacity, maxCapacity);
}

//UnpooledByteBufAllocator
protected ByteBuf newDirectBuffer(int initialCapacity, int maxCapacity) {
    ByteBuf buf;
    if (PlatformDependent.hasUnsafe()) {
        buf = new UnpooledUnsafeDirectByteBuf(this, initialCapacity, maxCapacity);
    } else {
        buf = new UnpooledDirectByteBuf(this, initialCapacity, maxCapacity);
    }
    return toLeakAwareBuffer(buf);
}

1234567891011121314151617181920212223

//AbstractByteBufAllocatorpublicByteBuf directBuffer(intinitialCapacity){returndirectBuffer(initialCapacity,Integer.MAX_VALUE);}publicByteBuf directBuffer(intinitialCapacity,intmaxCapacity){if(initialCapacity==0&&maxCapacity==0){returnemptyBuf;}validate(initialCapacity,maxCapacity);returnnewDirectBuffer(initialCapacity,maxCapacity);}//UnpooledByteBufAllocatorprotectedByteBuf newDirectBuffer(intinitialCapacity,intmaxCapacity){ByteBuf buf;if(PlatformDependent.hasUnsafe()){buf=newUnpooledUnsafeDirectByteBuf(this,initialCapacity,maxCapacity);}else{buf=newUnpooledDirectByteBuf(this,initialCapacity,maxCapacity);}returntoLeakAwareBuffer(buf);}

UnpooledUnsafeDirectByteBuf是如何實現快取管理的？對Nio的ByteBuffer進行了封裝，通過ByteBuffer的allocateDirect方法實現快取的申請。

protected UnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc, ByteBuffer initialBuffer, int maxCapacity) {
    //判斷邏輯已經忽略
   this.alloc = alloc;
   setByteBuffer(allocateDirect(initialCapacity));
}

protected ByteBuffer allocateDirect(int initialCapacity) {
    return ByteBuffer.allocateDirect(initialCapacity);
}

private void setByteBuffer(ByteBuffer buffer) {
    ByteBuffer oldBuffer = this.buffer;
    if (oldBuffer != null) {
        if (doNotFree) {
            doNotFree = false;
        } else {
            freeDirect(oldBuffer);
        }
    }
    this.buffer = buffer;
    memoryAddress = PlatformDependent.directBufferAddress(buffer);
    tmpNioBuf = null;
    capacity = buffer.remaining();
}

123456789101112131415161718192021222324

protectedUnpooledUnsafeDirectByteBuf(ByteBufAllocator alloc,ByteBuffer initialBuffer,intmaxCapacity){//判斷邏輯已經忽略this.alloc=alloc;setByteBuffer(allocateDirect(initialCapacity));}protectedByteBuffer allocateDirect(intinitialCapacity){returnByteBuffer.allocateDirect(initialCapacity);}privatevoidsetByteBuffer(ByteBuffer buffer){ByteBuffer oldBuffer=this.buffer;if(oldBuffer!=null){if(doNotFree){doNotFree=false;}else{freeDirect(oldBuffer);}}this.buffer=buffer;memoryAddress=PlatformDependent.directBufferAddress(buffer);

深入淺出Netty：read

深入淺出Netty：read

深入淺出Netty：write

深入淺出Netty：NioEventLoop

深入淺出Netty：服務啟動

深入淺出Netty：ChannelPipeline

深入淺出Netty：accept

深入淺出Netty記憶體管理：PoolSubpage

深入淺出Netty記憶體管理：PoolChunk

深入淺出Netty記憶體管理：PoolArena

深入淺出Netty記憶體管理：PoolChunkList

CentOS7使用ssh不能登錄，報錯：Read from socket failed: Connection reset by peer

深入淺出講解：php的socket通信

深入淺出CSS：Div（一）

下載-深入淺出Netty源碼剖析、Netty實戰高性能分布式RPC、NIO+Netty5各種RPC架構實戰演練三部曲視頻教程

pip安裝selenium報錯：Read timed out

深入淺出MyBatis：JDBC和MyBatis介紹

深入淺出MyBatis：反射和動態代理

深入淺出MyBatis：「映射器」全了解

深入淺出MyBatis：MyBatis與Spring集成及實用場景

第二章從鍵盤或文件中獲取標準輸入：read命令

深入淺出Netty：read

相關推薦