快捷導(dǎo)航

Netty分布式抽象編碼器MessageToByteEncoder邏輯分析

更新時(shí)間：2022年03月29日 12:15:08 作者：向南是個(gè)萬人迷

這篇文章主要介紹了Netty分布式抽象編碼器MessageToByteEncoder的抽象邏輯分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

MessageToByteEncoder

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

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

首先看MessageToByteEncoder的類聲明

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

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

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)前對(duì)象是否可處理

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

我們看if塊中

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

buf = allocateBuffer(ctx, cast, preferDirect) 這里進(jìn)行緩沖區(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();
    }
}

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

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

回到write方法中:

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

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

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

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

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

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

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

您可能感興趣的文章: