前文回顧：Netty分布式編碼器及寫數(shù)據(jù)事件處理

MessageToByteEncoder

同解碼器一樣, 編碼器中也有一個抽象類叫MessageToByteEncoder, 其中定義了編碼器的骨架方法, 具體編碼邏輯交給子類實現(xiàn)

解碼器同樣也是個handler, 將寫出的數(shù)據(jù)進行截取處理, 我們在學(xué)習(xí)pipeline中我們知道, 寫數(shù)據(jù)的時候會傳遞write事件, 傳遞過程中會調(diào)用handler的write方法, 所以編碼器碼器可以重寫write方法, 將數(shù)據(jù)編碼成二進制字節(jié)流然后再繼續(xù)傳遞write事件

首先看MessageToByteEncoder的類聲明

public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter{
    //省略類體
}

這里繼承ChannelOutboundHandlerAdapter, 說明是個outBoundhandler, 我們知道write事件是個outBound事件, 而outBound事件只能通過outBoundHandler進行傳輸

write事件傳播過程中要調(diào)用handler的write方法

我們跟到MessageToByteEncoder的write方法中:

public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
    ByteBuf buf = null;
    try {
        if (acceptOutboundMessage(msg)) {
            @SuppressWarnings("unchecked")
            I cast = (I) msg;
            buf = allocateBuffer(ctx, cast, preferDirect);
            try {
                encode(ctx, cast, buf);
            } finally {
                ReferenceCountUtil.release(cast);
            }

            if (buf.isReadable()) {
                ctx.write(buf, promise);
            } else {
                buf.release();
                ctx.write(Unpooled.EMPTY_BUFFER, promise);
            }
            buf = null;
        } else {
            ctx.write(msg, promise);
        }
    } catch (EncoderException e) {
        throw e;
    } catch (Throwable e) {
        throw new EncoderException(e);
    } finally {
        if (buf != null) {
            buf.release();
        }
    }
}

首先通過 if (acceptOutboundMessage(msg)) 判斷當(dāng)前對象是否可處理

如果可處理, 則進入if塊中的邏輯, 如果不能處理, 則進入else塊, 通過ctx.write(msg, promise)繼續(xù)傳遞write事件

我們看if塊中

 I cast = (I) msg 這里是強制類型轉(zhuǎn)換, 轉(zhuǎn)換成I類型, I類型是個泛型, 具體類型由用戶定義

 buf = allocateBuffer(ctx, cast, preferDirect) 這里進行緩沖區(qū)分配

跟到allocateBuffer方法中

protected ByteBuf allocateBuffer(ChannelHandlerContext ctx, @SuppressWarnings("unused") I msg, 
                           boolean preferDirect) throws Exception {
    if (preferDirect) {
        return ctx.alloc().ioBuffer();
    } else {
        return ctx.alloc().heapBuffer();
    }
}

這里會直接通過ctx的內(nèi)存分配器進行內(nèi)存分配, 通過判斷preferDirect來分配堆內(nèi)存或者堆外內(nèi)存, 默認(rèn)情況下是分配堆外內(nèi)存

有關(guān)內(nèi)存分配, 我們之前已經(jīng)做過相關(guān)的剖析

回到write方法中:

內(nèi)存分配結(jié)束之后會調(diào)用encode(ctx, cast, buf)方法進行編碼, 該類由子類實現(xiàn)

子類可以通過繼承該類, 重寫encode方法, 將參數(shù)對象cast編碼成字節(jié)寫入到傳入的ByteBuf中, 就完成了編碼工作

編碼完成后后, 會通過ReferenceCountUtil.release(cast)將cast對象釋放

 if (buf.isReadable()) 這里判斷buf是否有可讀字節(jié), 如果有可讀字節(jié), 則繼續(xù)傳遞write事件

如果沒有可讀字節(jié), 則將buf進行釋放, 繼續(xù)傳播write事件, 傳遞一個空的ByteBuf

最后將buf設(shè)置為空

以上就是有關(guān)抽象編碼器的抽象邏輯, 具體的編碼邏輯還需要其子類去做，更多關(guān)于Netty分布式抽象編碼器MessageToByteEncoder的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論