在日常的網路開發當中，協議決議都是必須的作業內容，Netty中雖然內置了基于長度、分隔符的編解碼器，但在大部分場景中我們使用的都是自定義協議，所以Netty提供了 MessageToByteEncoder<I> 與 ByteToMessageDecoder 兩個抽象類，通過繼承重寫其中的encode與decode方法實作私有協議的編解碼，這篇文章我們就對Netty中的自定義編解碼器進行實踐與分析，
一、編解碼器的使用
下面是MessageToByteEncoder與ByteToMessageDecoder使用的簡單示例，其中不涉及具體的協議編解碼，
創建一個sever端服務
EventLoopGroup bossGroup = new NioEventLoopGroup(1); EventLoopGroup workerGroup = new NioEventLoopGroup(); final CodecHandler codecHandler = new CodecHandler(); try { ServerBootstrap b = new ServerBootstrap(); b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 100) .handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() { @Override public void initChannel(SocketChannel ch) throws Exception { ChannelPipeline p = ch.pipeline(); if (sslCtx != null) { p.addLast(sslCtx.newHandler(ch.alloc())); } //添加編解碼handler p.addLast(new MessagePacketDecoder(),new MessagePacketEncoder()); //添加自定義handler p.addLast(codecHandler); } }); // Start the server. ChannelFuture f = b.bind(PORT).sync();
繼承MessageToByteEncoder并重寫encode方法，實作編碼功能
public class MessagePacketEncoder extends MessageToByteEncoder<byte[]> { @Override protected void encode(ChannelHandlerContext ctx, byte[] bytes, ByteBuf out) throws Exception { //進行具體的編碼處理 這里對位元組陣列進行列印 System.out.println("編碼器收到資料："+BytesUtils.toHexString(bytes)); //寫入并傳送資料 out.writeBytes(bytes); } }
繼承ByteToMessageDecoder 并重寫decode方法，實作解碼功能
public class MessagePacketDecoder extends ByteToMessageDecoder { @Override protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out){ try { if (buffer.readableBytes() > 0) { // 待處理的訊息包 byte[] bytesReady = new byte[buffer.readableBytes()]; buffer.readBytes(bytesReady); //進行具體的解碼處理 System.out.println("解碼器收到資料："+ByteUtils.toHexString(bytesReady)); //這里不做過多處理直接把收到的訊息放入鏈表中，并向后傳遞 out.add(bytesReady); } }catch(Exception ex) { } } }
實作自定義的訊息處理handler,到這里其實你拿到的已經是編解碼后的資料
public class CodecHandler extends ChannelInboundHandlerAdapter{ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) { System.out.println("CodecHandler收到資料："+ByteUtils.toHexString((byte[])msg)); byte[] sendBytes = new byte[] {0x7E,0x01,0x02,0x7e}; ctx.write(sendBytes); } @Override public void channelReadComplete(ChannelHandlerContext ctx) { ctx.flush(); } @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // Close the connection when an exception is raised. cause.printStackTrace(); ctx.close(); } }
運行一個客戶端模擬發送位元組0x01,0x02,看一下輸出的執行結果
解碼器收到資料：0102 CodecHandler收到資料：0102 編碼器收到資料：7E01027E
根據輸出的結果可以看到訊息的入站與出站會按照pipeline中自定義的順序傳遞，同時通過重寫encode與decode方法實作我們需要的具體協議編解碼操作，
二、原始碼分析
通過上面的例子可以看到MessageToByteEncoder<I>與ByteToMessageDecoder分別繼承了ChannelInboundHandlerAdapter與ChannelOutboundHandlerAdapter，所以它們也是channelHandler的具體實作,并在創建sever時被添加到pipeline中， 同時為了方便我們使用，netty在這兩個抽象類中內置與封裝了一些其操作；訊息的出站和入站會分別觸發write與channelRead事件方法，所以上面例子中我們重寫的encode與decode方法，也都是在父類的write與channelRead方法中被呼叫，下面我們就別從這兩個方法入手，對整個編解碼的流程進行梳理與分析，
1、MessageToByteEncoder
編碼需要操作的是出站資料，所以在MessageToByteEncoder的write方法中會呼叫我們重寫的encode具體實作， 把我們內部定義的訊息物體編碼為最終要發送的位元組流資料發送出去，
@Override public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { ByteBuf buf = null; try { if (acceptOutboundMessage(msg)) {//判斷傳入的msg與你定義的型別是否一致 @SuppressWarnings("unchecked") I cast = (I) msg;//轉為你定義的訊息型別 buf = allocateBuffer(ctx, cast, preferDirect);//包裝成一個ByteBuf try { encode(ctx, cast, buf);//傳入宣告的ByteBuf，執行具體編碼操作 } finally { /** * 如果你定義的型別就是ByteBuf 這里可以幫助你釋放資源，不需要在自己釋放 * 如果你定義的訊息型別中包含ByteBuf，這里是沒有作用，需要你自己主動釋放 */ ReferenceCountUtil.release(cast);//釋放你傳入的資源 } //發送buf if (buf.isReadable()) { ctx.write(buf, promise); } else { buf.release(); ctx.write(Unpooled.EMPTY_BUFFER, promise); } buf = null; } else { //型別不一致的話，就直接發送不再執行encode方法，所以這里要注意如果你傳遞的訊息與泛型型別不一致，其實是不會執行的 ctx.write(msg, promise); } } catch (EncoderException e) { throw e; } catch (Throwable e) { throw new EncoderException(e); } finally { if (buf != null) { buf.release();//釋放資源 } } }
MessageToByteEncoder的write方法要實作的功能還是比較簡單的，就是把你傳入的資料型別進行轉換和發送；這里有兩點需要注意：
- 一般情況下，需要通過重寫encode方法把定義的泛型型別轉換為ByteBuf型別， write方法內部自動幫你執行傳遞或發送操作；
- 代碼中雖然有通過ReferenceCountUtil.release(cast)釋放你定義的型別資源，但如果定義的訊息類中包含ByteBuf物件，仍需要主動釋放該物件資源；
2、ByteToMessageDecoder
從命名上就可以看出ByteToMessageDecoder解碼器的作用是把位元組流資料編碼轉換為我們需要的資料格式
作為入站事件，解碼操作的入口自然是channelRead方法
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { if (msg instanceof ByteBuf) {//如果訊息是bytebuff CodecOutputList out = CodecOutputList.newInstance();//實體化一個鏈表 try { ByteBuf data = (ByteBuf) msg; first = cumulation == null; if (first) { cumulation = data; } else { cumulation = cumulator.cumulate(ctx.alloc(), cumulation, data); } callDecode(ctx, cumulation, out);//開始解碼 } catch (DecoderException e) { throw e; } catch (Exception e) { throw new DecoderException(e); } finally { if (cumulation != null && !cumulation.isReadable()) {//不為空且沒有可讀資料，釋放資源 numReads = 0; cumulation.release(); cumulation = null; } else if (++ numReads >= discardAfterReads) { // We did enough reads already try to discard some bytes so we not risk to see a OOME. // See https://github.com/netty/netty/issues/4275 numReads = 0; discardSomeReadBytes(); } int size = out.size(); decodeWasNull = !out.insertSinceRecycled(); fireChannelRead(ctx, out, size);//向下傳遞訊息 out.recycle(); } } else { ctx.fireChannelRead(msg); } }
callDecode方法內部通過while回圈的方式對ByteBuf資料進行解碼，直到其中沒有可讀資料
protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) { try { while (in.isReadable()) {//判斷ByteBuf是還有可讀資料 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; } int oldInputLength = in.readableBytes(); decodeRemovalReentryProtection(ctx, in, out);//開始呼叫decode方法 // 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; } //這里如果鏈表為空且bytebuf沒有可讀資料，就跳出回圈 if (outSize == out.size()) { if (oldInputLength == in.readableBytes()) { break; } else {//有可讀資料繼續讀取 continue; } } if (oldInputLength == in.readableBytes()) {//beytebuf沒有讀取，但卻進行了解碼 throw new DecoderException( StringUtil.simpleClassName(getClass()) + ".decode() did not read anything but decoded a message."); } if (isSingleDecode()) {//是否設定了每條入站資料只解碼一次，默認false break; } } } catch (DecoderException e) { throw e; } catch (Exception cause) { throw new DecoderException(cause); } }
decodeRemovalReentryProtection方法內部會呼叫我們重寫的decode解碼實作
final void decodeRemovalReentryProtection(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { decodeState = STATE_CALLING_CHILD_DECODE;//標記狀態 try { decode(ctx, in, out);//呼叫我們重寫的decode解碼實作 } finally { boolean removePending = decodeState == STATE_HANDLER_REMOVED_PENDING; decodeState = STATE_INIT; if (removePending) {//這里判斷標記，防止handlerRemoved事件與解碼操作沖突 handlerRemoved(ctx); } } }
channelRead方法中接受到資料經過一系列邏輯處理，最侄訓呼叫我們重寫的decode方法實作具體的解碼功能；在decode方法中我們只需要ByteBuf型別的資料決議為我們需要的資料格式直接放入 List<Object> out鏈表中即可，ByteToMessageDecoder會自動幫你向下傳遞訊息，
三、總結
通過上面的講解，我們可以對Netty中內置自定義編解碼器MessageToByteEncoder與ByteToMessageDecoder有一定的了解，其實它們本質上是Netty封裝的一組專門用于自定義編解碼的channelHandler實作類，在實際開發當中基于這兩個抽象類的實作非常具有實用性，所以在這里稍作分析， 其中如有不足與不正確的地方還望指出與海涵，
關注微信公眾號，查看更多技術文章，
轉載說明：未經授權不得轉載，授權后務必注明來源（注明：來源于公眾號：架構空間， 作者：大凡）
轉載請註明出處，本文鏈接：https://www.uj5u.com/houduan/158649.html
標籤：Java