快捷導(dǎo)航

Netty如何自定義編碼解碼器

更新時間：2025年03月06日 10:35:31 作者：帥氣的人123

Netty自定義編碼解碼器：InboundHandler處理入棧數(shù)據(jù),OutboundHandler處理出棧數(shù)據(jù),解碼器繼承ByteToMessageDecoder,編碼器繼承MessageToByteEncoder,ReplayingDecoder簡化了解碼邏輯,但可能因異常重試導(dǎo)致性能下降

Netty 自定義編碼解碼器

入棧：InboundHandler ，出棧：OutboundHandler。Netty 構(gòu)建 chain 來處理業(yè)務(wù)。

自定義一個解碼器

解碼器主要是對客戶端發(fā)送的消息進(jìn)行解碼處理，所以他是一個入棧的處理器，因此他會有一個入棧的標(biāo)識ibound=true;

解碼器一般我都們都會基層 Netty 提供給的一個實(shí)現(xiàn)類來實(shí)現(xiàn)自己的解碼邏輯 -> io.netty.handler.codec.ByteToMessageDecoder 這就是解碼的抽象類，默認(rèn)我們要實(shí)現(xiàn)一個抽象方法

com.netty.codec.custom.InboundAndOutboundHandler#decode

這個方法有大概三個參數(shù)；

ChannelHandlerContext channelHandlerContext 這個是上下文信息，可以獲取通道、管道等信息。
ByteBuf byteBuf 客戶端發(fā)送的消息就是存在這個參數(shù)的對象里面我們要通過這個對象的read***方法讀取我們需要的數(shù)據(jù)類型，可以是 Long，Byte 等類型的數(shù)據(jù)然后我們可以就可以轉(zhuǎn)換成為我們需要的格式。
List<Object> list 集合，將解碼后的數(shù)據(jù)傳遞給下一個inboundHandler 處理類。

代碼示例

@Override
protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) {
    // Long => 8 byte
    if (byteBuf.readableBytes() >= 8) {
        list.add(byteBuf.readLong());
    } else log.warn("字節(jié)數(shù)異常 => {}", byteBuf.readableBytes());
}

這樣我們就實(shí)現(xiàn)了一個簡單的解碼器。

自定義一個編碼器

解碼器主要是對服務(wù)端將要發(fā)送給客戶端的消息進(jìn)行編碼處理，所以他是一個出棧的處理器，因此他會有一個入棧的標(biāo)識outbound=true;

使用 Netty 提供的抽象類 => io.netty.handler.codec.MessageToByteEncoder<T> 泛型 T 表示你要發(fā)送的消息的類型，實(shí)現(xiàn)抽象方法 => com.netty.codec.custom.OutboundHandler#encode 方法的參數(shù)有三個：

ChannelHandlerContext channelHandlerContext 這個是上下文信息，可以獲取通道、管道等信息。
Long msg 服務(wù)端要發(fā)送給客戶端的消息
ByteBuf byteBuf

代碼示例

@Override
protected void encode(ChannelHandlerContext channelHandlerContext, Long msg, ByteBuf byteBuf) {
    byteBuf.writeLong(msg);
    log.info("發(fā)送消息成功");
}

后續(xù) -> io.netty.handler.codec.ReplayingDecoder

ByteToMessage 其實(shí)在使用過程中會遇到一些問題，例如：

當(dāng)我們的解碼器中想要將字節(jié)轉(zhuǎn)換為一個 Integer ，我們知道 Integer 是四個字節(jié)的，但是如果在讀取的時候不夠四個字節(jié)，這個時候我們就需要做一些判斷邏輯 => if (byteBuf.readableBytes() >= 4) 當(dāng)這個返回值為 true 的時候我么就可以繼續(xù)執(zhí)行解碼的邏輯。

那我們怎么可以跳過這一步不判斷直接進(jìn)行我們的轉(zhuǎn)換邏輯呢？

這個時候就可以使用 Netty 的

io.netty.handler.codec.ReplayingDecoder

可以不用判斷可讀字節(jié)數(shù)的原理：

ReplayingDecoder 是 ByteToMessage 的子類，源碼如下：

public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder {
    ...
}

ReplayingDecoder 的秘密就是對 ByteToMessage 的 CallDecode(...) 方法的重寫，觀摩一下具體實(shí)現(xiàn)：

@Override
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    replayable.setCumulation(in);
    try {
        while (in.isReadable()) {
            int oldReaderIndex = checkpoint = in.readerIndex();
            int outSize = out.size();

            if (outSize > 0) {
                fireChannelRead(ctx, out, outSize);
                out.clear();

                // Check if this handler was removed before continuing with decoding.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See:
                // - https://github.com/netty/netty/issues/4635
                if (ctx.isRemoved()) {
                    break;
                }
                outSize = 0;
            }

            S oldState = state;
            int oldInputLength = in.readableBytes();
            try {
                decodeRemovalReentryProtection(ctx, replayable, out);

                // Check if this handler was removed before continuing the loop.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See https://github.com/netty/netty/issues/1664
                if (ctx.isRemoved()) {
                    break;
                }

                if (outSize == out.size()) {
                    if (oldInputLength == in.readableBytes() && oldState == state) {
                        throw new DecoderException(
                                StringUtil.simpleClassName(getClass()) + ".decode() must consume the inbound " +
                                "data or change its state if it did not decode anything.");
                    } else {
                        // Previous data has been discarded or caused state transition.
                        // Probably it is reading on.
                        continue;
                    }
                }
            } catch (Signal replay) {
                replay.expect(REPLAY);

                // Check if this handler was removed before continuing the loop.
                // If it was removed, it is not safe to continue to operate on the buffer.
                //
                // See https://github.com/netty/netty/issues/1664
                if (ctx.isRemoved()) {
                    break;
                }

                // Return to the checkpoint (or oldPosition) and retry.
                int checkpoint = this.checkpoint;
                if (checkpoint >= 0) {
                    in.readerIndex(checkpoint);
                } else {
                    // Called by cleanup() - no need to maintain the readerIndex
                    // anymore because the buffer has been released already.
                }
                break;
            }

            if (oldReaderIndex == in.readerIndex() && oldState == state) {
                throw new DecoderException(
                       StringUtil.simpleClassName(getClass()) + ".decode() method must consume the inbound data " +
                       "or change its state if it decoded something.");
            }
            if (isSingleDecode()) {
                break;
            }
        }
    } catch (DecoderException e) {
        throw e;
    } catch (Exception cause) {
        throw new DecoderException(cause);
    }
}

實(shí)現(xiàn)不需要判斷的邏輯就是因為

int oldReaderIndex = checkpoint = in.readerIndex();

如果在執(zhí)行過程中出現(xiàn)異常就會在代碼中重置 index

總結(jié)

雖然 ReplayingDecoder節(jié)約了判斷的邏輯，但是從他的代碼實(shí)現(xiàn)邏輯看到是通過拋出異常來不斷的重試，所以在某些特殊的情況下會造成性能的下降。所以還是再選擇的時候要根據(jù)自己的實(shí)際需求來判斷是使用 ByteToMessage 還是使用 ReplayingDecoder。

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

您可能感興趣的文章: