---

highlight: arduino-light

Netty內(nèi)置的解碼器

在前兩節(jié)課我們介紹了 TCP 拆包/粘包的問題，以及如何使用 Netty 實現(xiàn)自定義協(xié)議的編解碼?？梢钥吹?，網(wǎng)絡(luò)通信的底層實現(xiàn)，Netty 都已經(jīng)幫我們封裝好了，我們只需要擴展 ChannelHandler 實現(xiàn)自定義的編解碼邏輯即可。

更加人性化的是，Netty 提供了很多開箱即用的解碼器，這些解碼器基本覆蓋了 TCP 拆包/粘包的通用解決方案。本節(jié)課我們將對 Netty 常用的解碼器進行講解，一起探索下它們有哪些用法和技巧。

在本節(jié)課開始之前，我們首先回顧一下 TCP 拆包/粘包的主流解決方案。并梳理出 Netty 對應(yīng)的編碼器類。

定長:FixedLengthFrameDecoder

固定長度解碼器 FixedLengthFrameDecoder 非常簡單，直接通過構(gòu)造函數(shù)設(shè)置固定長度的大小 frameLength，無論接收方一次獲取多大的數(shù)據(jù)，都會嚴格按照 frameLength 進行解碼。如果累積讀取到長度大小為 frameLength 的消息，那么解碼器認為已經(jīng)獲取到了一個完整的消息。如果消息長度小于 frameLength，F(xiàn)ixedLengthFrameDecoder 解碼器會一直等后續(xù)數(shù)據(jù)包的到達，直至獲得完整的消息。下面我們通過一個例子感受一下使用 Netty 實現(xiàn)固定長度解碼是多么簡單。

java package io.netty.example.decode; ? import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioChannelOption; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.FixedLengthFrameDecoder; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.util.CharsetUtil; ? /** * Echoes back any received data from a client. */ public final class EchoServer { ? ? ?public static void main(String[] args) throws Exception { ? ? ? ? ?EventLoopGroup workerGroup = new NioEventLoopGroup(); ? ? ? ?EventLoopGroup bossGroup = new NioEventLoopGroup(); ? ? ? ?final EchoServerHandler serverHandler = new EchoServerHandler(); ? ? ? ?try { ? ? ? ? ? ?ServerBootstrap b = new ServerBootstrap(); ? ? ? ? ? ? ?b.group(bossGroup,workerGroup) ? ? ? ? ? ? ?//通過反射創(chuàng)建反射工廠類根據(jù)無參構(gòu)造函數(shù) 反射生成實例 ? ? ? ? ? ? ?//將NioServerSocketChannel綁定到了bossGroup ? ? ? ? ? ? ? ? ? ?//NioServerSocketChannel接收到請求會創(chuàng)建SocketChannel放入workerGroup ? ? ? ? ? ? .channel(NioServerSocketChannel.class) ? ? ? ? ? ? ? ? ? ? ?//指的是SocketChannel ? ? ? ? ? ? .childOption(ChannelOption.SO_KEEPALIVE,true) ? ? ? ? ? ? ? ? ? ?//指的是SocketChannel ? ? ? ? ? ? .childOption(NioChannelOption.SO_KEEPALIVE,Boolean.TRUE) ? ? ? ? ? ? ? ? ? ?//默認不使用堆外內(nèi)存 ? ? ? ? ? ? ? ? ? .childOption(ChannelOption.ALLOCATOR,PooledByteBufAllocator.DEFAULT) ? ? ? ? ? ? ? ? ? ?//false 不使用堆外內(nèi)存 ? ? ? ? ? ? ? ? ? .childOption(ChannelOption.ALLOCATOR,new UnpooledByteBufAllocator(false)) ? ? ? ? ?// ? .handler(new LoggingHandler(LogLevel.INFO)) ? ? ? ? ? ? .childHandler(new ChannelInitializer<SocketChannel>() { ? ? ? ? ? ? ? ? @Override ? ? ? ? ? ? ? ? public void initChannel(SocketChannel ch) throws Exception { ? ? ? ? ? ? ? ? ? ? ChannelPipeline p = ch.pipeline(); ? ? ? ? ? ? ? ? ? ?// p.addLast(new LoggingHandler(LogLevel.INFO)); ? ? ? ? ? ? ? ? ? ? ch.pipeline().addLast(new FixedLengthFrameDecoder(10)); ? ? ? ? ? ? ? ? ? ? p.addLast(serverHandler); ? ? ? ? ? ? ? ? } ? ? ? ? ? ? }); ? ? ? ? ? ? ?ChannelFuture f = b.bind(8090).sync(); ? ? ? ? ? ?f.channel().closeFuture().sync(); ? ? ? } finally { ? ? ? ? ? ?workerGroup.shutdownGracefully(); ? ? ? } ? } } ? @ChannelHandler.Sharable class EchoServerHandler extends ChannelInboundHandlerAdapter { ? ?@Override ? ?public void channelRead(ChannelHandlerContext ctx, Object msg) { ? ? ? ?System.out.println("Receive client : [" + ((ByteBuf) msg).toString(CharsetUtil.UTF_8) + "]"); ? } }

java package io.netty.example.decode; ? import io.netty.bootstrap.Bootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; ? ? public final class EchoClient { ? ? ?public static void main(String[] args) throws Exception { ? ? ? ?// Configure the client. ? ? ? ?EventLoopGroup group = new NioEventLoopGroup(); ? ? ? ?try { ? ? ? ? ? ?Bootstrap b = new Bootstrap(); ? ? ? ? ? ?b.group(group) ? ? ? ? ? ? .channel(NioSocketChannel.class) ? ? ? ? ? ? .option(ChannelOption.TCP_NODELAY, true) ? ? ? ? ? ? .handler(new ChannelInitializer<SocketChannel>() { ? ? ? ? ? ? ? ? @Override ? ? ? ? ? ? ? ? public void initChannel(SocketChannel ch) throws Exception { ? ? ? ? ? ? ? ? ? ? ChannelPipeline p = ch.pipeline(); ? ? ? ? ? ? ? ? ? // p.addLast(new LoggingHandler(LogLevel.INFO)); ? ? ? ? ? ? ? ? ? ? p.addLast(new EchoClientHandler()); ? ? ? ? ? ? ? ? } ? ? ? ? ? ? }); ? ? ? ? ? ? ?// Start the client. ? ? ? ? ? ?ChannelFuture f = b.connect("127.0.0.1", 8090).sync(); ? ? ? ? ? ? ?// Wait until the connection is closed. ? ? ? ? ? ?f.channel().closeFuture().sync(); ? ? ? } finally { ? ? ? ? ? ?// Shut down the event loop to terminate all threads. ? ? ? ? ? ?group.shutdownGracefully(); ? ? ? } ? } } ?

java package io.netty.example.decode; ? import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; ? import java.util.concurrent.TimeUnit; ? /** * Handler implementation for the echo client. It initiates the ping-pong * traffic between the echo client and server by sending the first message to * the server. */ public class EchoClientHandler extends ChannelInboundHandlerAdapter { ? ? ?private final ByteBuf firstMessage; ? ? ?/** ? ? * Creates a client-side handler. ? ? */ ? ?//TODO 修改1234567890 看看10位數(shù)字 和 非10位數(shù)字的區(qū)別 ? ?public EchoClientHandler() { ? ? ? ?firstMessage = Unpooled.wrappedBuffer("1234567890".getBytes()); ? } ? ? ?@Override ? ?public void channelActive(ChannelHandlerContext ctx) { ? ? ? ?System.out.println("客戶端發(fā)送消息" + firstMessage.toString()); ? ? ? ?ctx.writeAndFlush(firstMessage); ? } ? ? ?@Override ? ?public void channelRead(ChannelHandlerContext ctx, Object msg) { ? ? ? // ctx.write(msg); ? } ? ? ?@Override ? ?public void channelReadComplete(ChannelHandlerContext ctx) throws InterruptedException { ? ? ? ?TimeUnit.SECONDS.sleep(3); ? ? ? ?ctx.flush(); ? } ? ? ?@Override ? ?public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ? ? ? // cause.printStackTrace(); ? ? ? ?ctx.close(); ? } } ?

在上述服務(wù)端的代碼中使用了固定 10 字節(jié)的解碼器，并在解碼之后通過 EchoServerHandler 打印結(jié)果。我們可以啟動服務(wù)端，通過 telnet 命令像服務(wù)端發(fā)送數(shù)據(jù)，觀察代碼輸出的結(jié)果。

java telnet localhost 8088 Trying ::1... Connected to localhost. Escape character is '^]'. 1234567890123 456789012

按10個字節(jié)一組進行解析注意有個換行符

服務(wù)端輸出：

java Receive client : [1234567890] Receive client : [123 45678]

分隔:DelimiterBasedFrameDecoder

java public class DelimiterBasedFrameDecoder extends ByteToMessageDecoder { ? ? ?private final ByteBuf[] delimiters; ? ?private final int maxFrameLength; ? ?private final boolean stripDelimiter; ? ?private final boolean failFast; ? ?private boolean discardingTooLongFrame; ? ?private int tooLongFrameLength; ? ?/** Set only when decoding with "\n" and "\r\n" as the delimiter. */ ? ?private final LineBasedFrameDecoder lineBasedDecoder; ? ? ?/** ? ? * Creates a new instance. ? ? * ? ? * @param maxFrameLength the maximum length of the decoded frame. ? ? * ? ? ? ? ? ? ? ? ? ? ? A {@link TooLongFrameException} is thrown if ? ? * ? ? ? ? ? ? ? ? ? ? ? the length of the frame exceeds this value. ? ? * @param delimiter the delimiter ? ? */ ? ?public DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf delimiter) { ? ? ? ?this(maxFrameLength, true, delimiter); ? } ? ? ?/** ? ? * Creates a new instance. ? ? * ? ? * @param maxFrameLength the maximum length of the decoded frame. ? ? * ? ? ? ? ? ? ? ? ? ? ? A {@link TooLongFrameException} is thrown if ? ? * ? ? ? ? ? ? ? ? ? ? ? the length of the frame exceeds this value. ? ? * @param stripDelimiter whether the decoded frame should strip out the ? ? * ? ? ? ? ? ? ? ? ? ? ? delimiter or not ? ? * @param delimiter the delimiter ? ? */ ? ?public DelimiterBasedFrameDecoder( ? ? ? ? ? ?int maxFrameLength, boolean stripDelimiter, ByteBuf delimiter) { ? ? ? ?this(maxFrameLength, stripDelimiter, true, delimiter); ? } ? ? ?/** ? ? * Creates a new instance. ? ? * ? ? * @param maxFrameLength the maximum length of the decoded frame. ? ? * ? ? ? ? ? ? ? ? ? ? ? A {@link TooLongFrameException} is thrown if ? ? * ? ? ? ? ? ? ? ? ? ? ? the length of the frame exceeds this value. ? ? * @param stripDelimiter whether the decoded frame should strip out the ? ? * ? ? ? ? ? ? ? ? ? ? ? delimiter or not ? ? * @param failFast If <tt>true</tt>, a {@link TooLongFrameException} is ? ? * ? ? ? ? ? ? ? ? thrown as soon as the decoder notices the length of the ? ? * ? ? ? ? ? ? ? ? frame will exceed <tt>maxFrameLength</tt> regardless of ? ? * ? ? ? ? ? ? ? ? whether the entire frame has been read. ? ? * ? ? ? ? ? ? ? ? If <tt>false</tt>, a {@link TooLongFrameException} is ? ? * ? ? ? ? ? ? ? ? thrown after the entire frame that exceeds ? ? * ? ? ? ? ? ? ? ? <tt>maxFrameLength</tt> has been read. ? ? * @param delimiter the delimiter ? ? */ ? ?public DelimiterBasedFrameDecoder( ? ? ? ? ? ?int maxFrameLength, boolean stripDelimiter, boolean failFast, ? ? ? ? ? ?ByteBuf delimiter) { ? ? ? ?this(maxFrameLength, stripDelimiter, failFast, new ByteBuf[] { ? ? ? ? ? ? ? ?delimiter.slice(delimiter.readerIndex(), delimiter.readableBytes())}); ? } ? ? ?/** ? ? * Creates a new instance. ? ? * ? ? * @param maxFrameLength the maximum length of the decoded frame. ? ? * ? ? ? ? ? ? ? ? ? ? ? A {@link TooLongFrameException} is thrown if ? ? * ? ? ? ? ? ? ? ? ? ? ? the length of the frame exceeds this value. ? ? * @param delimiters the delimiters ? ? */ ? ?public DelimiterBasedFrameDecoder(int maxFrameLength, ByteBuf... delimiters) { ? ? ? ?this(maxFrameLength, true, delimiters); ? } ? ? ?/** ? ? * Creates a new instance. ? ? * ? ? * @param maxFrameLength the maximum length of the decoded frame. ? ? * ? ? ? ? ? ? ? ? ? ? ? A {@link TooLongFrameException} is thrown if ? ? * ? ? ? ? ? ? ? ? ? ? ? the length of the frame exceeds this value. ? ? * @param stripDelimiter whether the decoded frame should strip out the ? ? * ? ? ? ? ? ? ? ? ? ? ? delimiter or not ? ? * @param delimiters the delimiters ? ? */ ? ?public DelimiterBasedFrameDecoder( ? ? ? ? ? ?int maxFrameLength, boolean stripDelimiter, ByteBuf... delimiters) { ? ? ? ?this(maxFrameLength, stripDelimiter, true, delimiters); ? } ? ? ?/** ? ? * Creates a new instance. ? ? * ? ? * @param maxFrameLength the maximum length of the decoded frame. ? ? * ? ? ? ? ? ? ? ? ? ? ? A {@link TooLongFrameException} is thrown if ? ? * ? ? ? ? ? ? ? ? ? ? ? the length of the frame exceeds this value. ? ? * @param stripDelimiter whether the decoded frame should strip out the ? ? * ? ? ? ? ? ? ? ? ? ? ? delimiter or not ? ? * @param failFast If <tt>true</tt>, a {@link TooLongFrameException} is ? ? * ? ? ? ? ? ? ? ? thrown as soon as the decoder notices the length of the ? ? * ? ? ? ? ? ? ? ? frame will exceed <tt>maxFrameLength</tt> regardless of ? ? * ? ? ? ? ? ? ? ? whether the entire frame has been read. ? ? * ? ? ? ? ? ? ? ? If <tt>false</tt>, a {@link TooLongFrameException} is ? ? * ? ? ? ? ? ? ? ? thrown after the entire frame that exceeds ? ? * ? ? ? ? ? ? ? ? <tt>maxFrameLength</tt> has been read. ? ? * @param delimiters the delimiters ? ? */ ? ?public DelimiterBasedFrameDecoder( ? ? ? ? ? ?int maxFrameLength, boolean stripDelimiter, boolean failFast, ByteBuf... delimiters) { ? ? ? ?validateMaxFrameLength(maxFrameLength); ? ? ? ?if (delimiters == null) { ? ? ? ? ? ?throw new NullPointerException("delimiters"); ? ? ? } ? ? ? ?if (delimiters.length == 0) { ? ? ? ? ? ?throw new IllegalArgumentException("empty delimiters"); ? ? ? } ? ? ? ? ?if (isLineBased(delimiters) && !isSubclass()) { ? ? ? ? ? ?lineBasedDecoder = new LineBasedFrameDecoder(maxFrameLength, stripDelimiter, failFast); ? ? ? ? ? ?this.delimiters = null; ? ? ? } else { ? ? ? ? ? ?this.delimiters = new ByteBuf[delimiters.length]; ? ? ? ? ? ?for (int i = 0; i < delimiters.length; i ++) { ? ? ? ? ? ? ? ?ByteBuf d = delimiters[i]; ? ? ? ? ? ? ? ?validateDelimiter(d); ? ? ? ? ? ? ? ?this.delimiters[i] = d.slice(d.readerIndex(), d.readableBytes()); ? ? ? ? ? } ? ? ? ? ? ?lineBasedDecoder = null; ? ? ? } ? ? ? ?this.maxFrameLength = maxFrameLength; ? ? ? ?this.stripDelimiter = stripDelimiter; ? ? ? ?this.failFast = failFast; ? } ? ? ?/** Returns true if the delimiters are "\n" and "\r\n". */ ? ?private static boolean isLineBased(final ByteBuf[] delimiters) { ? ? ? ?if (delimiters.length != 2) { ? ? ? ? ? ?return false; ? ? ? } ? ? ? ?ByteBuf a = delimiters[0]; ? ? ? ?ByteBuf b = delimiters[1]; ? ? ? ?if (a.capacity() < b.capacity()) { ? ? ? ? ? ?a = delimiters[1]; ? ? ? ? ? ?b = delimiters[0]; ? ? ? } ? ? ? ?return a.capacity() == 2 && b.capacity() == 1 ? ? ? ? ? ? ? ?&& a.getByte(0) == '\r' && a.getByte(1) == '\n' ? ? ? ? ? ? ? ?&& b.getByte(0) == '\n'; ? } ? ? ?/** ? ? * Return {@code true} if the current instance is a subclass of DelimiterBasedFrameDecoder ? ? */ ? ?private boolean isSubclass() { ? ? ? ?return getClass() != DelimiterBasedFrameDecoder.class; ? } ? ? ?@Override ? ?protected final void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { ? ? ? ?Object decoded = decode(ctx, in); ? ? ? ?if (decoded != null) { ? ? ? ? ? ?out.add(decoded); ? ? ? } ? } ? ? ? ?protected Object decode(ChannelHandlerContext ctx, ByteBuf buffer) throws Exception { ? ? ? ?if (lineBasedDecoder != null) { ? ? ? ? ? ?return lineBasedDecoder.decode(ctx, buffer); ? ? ? } ? ? ? ?// Try all delimiters and choose the delimiter which yields the shortest frame. ? ? ? ?int minFrameLength = Integer.MAX_VALUE; ? ? ? ?ByteBuf minDelim = null; ? ? ? ?for (ByteBuf delim: delimiters) { ? ? ? ? ? ?int frameLength = indexOf(buffer, delim); ? ? ? ? ? ?if (frameLength >= 0 && frameLength < minFrameLength) { ? ? ? ? ? ? ? ?minFrameLength = frameLength; ? ? ? ? ? ? ? ?minDelim = delim; ? ? ? ? ? } ? ? ? } ? ? ? ? ?if (minDelim != null) { ? ? ? ? ? ?int minDelimLength = minDelim.capacity(); ? ? ? ? ? ?ByteBuf frame; ? ? ? ? ? ? ?if (discardingTooLongFrame) { ? ? ? ? ? ? ? ?// We've just finished discarding a very large frame. ? ? ? ? ? ? ? ?// Go back to the initial state. ? ? ? ? ? ? ? ?discardingTooLongFrame = false; ? ? ? ? ? ? ? ?buffer.skipBytes(minFrameLength + minDelimLength); ? ? ? ? ? ? ? ? ?int tooLongFrameLength = this.tooLongFrameLength; ? ? ? ? ? ? ? ?this.tooLongFrameLength = 0; ? ? ? ? ? ? ? ?if (!failFast) { ? ? ? ? ? ? ? ? ? ?fail(tooLongFrameLength); ? ? ? ? ? ? ? } ? ? ? ? ? ? ? ?return null; ? ? ? ? ? } ? ? ? ? ? ? ?if (minFrameLength > maxFrameLength) { ? ? ? ? ? ? ? ?// Discard read frame. ? ? ? ? ? ? ? ?buffer.skipBytes(minFrameLength + minDelimLength); ? ? ? ? ? ? ? ?fail(minFrameLength); ? ? ? ? ? ? ? ?return null; ? ? ? ? ? } ? ? ? ? ? ? ?if (stripDelimiter) { ? ? ? ? ? ? ? ?frame = buffer.readRetainedSlice(minFrameLength); ? ? ? ? ? ? ? ?buffer.skipBytes(minDelimLength); ? ? ? ? ? } else { ? ? ? ? ? ? ? ?frame = buffer.readRetainedSlice(minFrameLength + minDelimLength); ? ? ? ? ? } ? ? ? ? ? ? ?return frame; ? ? ? } else { ? ? ? ? ? ?if (!discardingTooLongFrame) { ? ? ? ? ? ? ? ?if (buffer.readableBytes() > maxFrameLength) { ? ? ? ? ? ? ? ? ? ?// Discard the content of the buffer until a delimiter is found. ? ? ? ? ? ? ? ? ? ?tooLongFrameLength = buffer.readableBytes(); ? ? ? ? ? ? ? ? ? ?buffer.skipBytes(buffer.readableBytes()); ? ? ? ? ? ? ? ? ? ?discardingTooLongFrame = true; ? ? ? ? ? ? ? ? ? ?if (failFast) { ? ? ? ? ? ? ? ? ? ? ? ?fail(tooLongFrameLength); ? ? ? ? ? ? ? ? ? } ? ? ? ? ? ? ? } ? ? ? ? ? } else { ? ? ? ? ? ? ? ?// Still discarding the buffer since a delimiter is not found. ? ? ? ? ? ? ? ?tooLongFrameLength += buffer.readableBytes(); ? ? ? ? ? ? ? ?buffer.skipBytes(buffer.readableBytes()); ? ? ? ? ? } ? ? ? ? ? ?return null; ? ? ? } ? } ? ? } ?

使用特殊分隔符解碼器 DelimiterBasedFrameDecoder 之前我們需要了解以下幾個屬性的作用。

• delimiters

delimiters 指定特殊分隔符，通過寫入 ByteBuf 作為參數(shù)傳入。delimiters 的類型是 ByteBuf 數(shù)組，所以我們可以同時指定多個分隔符，但是最終會選擇長度最短的分隔符進行消息拆分。

例如接收方收到的數(shù)據(jù)為：

java +--------------+ | ABC\nDEF\r\n | +--------------+

如果指定的多個分隔符為 \n 和 \r\n，DelimiterBasedFrameDecoder 會退化成使用 LineBasedFrameDecoder 進行解析，那么會解碼出兩個消息。

java +-----+-----+ | ABC | DEF | +-----+-----+

如果指定的特定分隔符只有 \r\n，那么只會解碼出一個消息：

java +----------+ | ABC\nDEF | +----------+

• maxLength

maxLength 是報文最大長度的限制。如果超過 maxLength 還沒有檢測到指定分隔符，將會拋出 TooLongFrameException?？梢哉f maxLength 是對程序在極端情況下的一種保護措施。

• failFast

failFast 與 maxLength 需要搭配使用，通過設(shè)置 failFast 可以控制拋出 TooLongFrameException 的時機，可以說 Netty 在細節(jié)上考慮得面面俱到。如果 failFast=true，那么在超出 maxLength 會立即拋出 TooLongFrameException，不再繼續(xù)進行解碼。如果 failFast=false，那么會等到解碼出一個完整的消息后才會拋出 TooLongFrameException。

• stripDelimiter

stripDelimiter 的作用是判斷解碼后得到的消息是否去除分隔符。如果 stripDelimiter=false，特定分隔符為 \n，那么上述數(shù)據(jù)包解碼出的結(jié)果為：

java +-------+---------+ | ABC\n | DEF\r\n | +-------+---------+

下面我們還是結(jié)合代碼示例學習 DelimiterBasedFrameDecoder 的用法，依然以固定編碼器小節(jié)中使用的代碼為基礎(chǔ)稍做改動，引入特殊分隔符解碼器 DelimiterBasedFrameDecoder：

java package io.netty.example.decode; ? import io.netty.bootstrap.ServerBootstrap; import io.netty.buffer.ByteBuf; import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.Unpooled; import io.netty.buffer.UnpooledByteBufAllocator; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioChannelOption; import io.netty.channel.socket.nio.NioServerSocketChannel; import io.netty.handler.codec.DelimiterBasedFrameDecoder; import io.netty.handler.codec.FixedLengthFrameDecoder; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; import io.netty.util.CharsetUtil; ? /** * Echoes back any received data from a client. */ public final class EchoServer { ? ? ?public static void main(String[] args) throws Exception { ? ? ? ? ?EventLoopGroup workerGroup = new NioEventLoopGroup(); ? ? ? ?EventLoopGroup bossGroup = new NioEventLoopGroup(); ? ? ? ?final EchoServerHandler serverHandler = new EchoServerHandler(); ? ? ? ?try { ? ? ? ? ? ?ServerBootstrap b = new ServerBootstrap(); ? ? ? ? ? ? ?b.group(bossGroup, workerGroup) ? ? ? ? ? ? ? ? ? ?//通過反射創(chuàng)建反射工廠類根據(jù)無參構(gòu)造函數(shù) 反射生成實例 ? ? ? ? ? ? ? ? ? ?//將NioServerSocketChannel綁定到了bossGroup ? ? ? ? ? ? ? ? ? ?//NioServerSocketChannel接收到請求會創(chuàng)建SocketChannel放入workerGroup ? ? ? ? ? ? ? ? ? .channel(NioServerSocketChannel.class) ? ? ? ? ? ? ? ? ? ? ?//指的是SocketChannel ? ? ? ? ? ? ? ? ? .childOption(ChannelOption.SO_KEEPALIVE, true) ? ? ? ? ? ? ? ? ? ?//指的是SocketChannel ? ? ? ? ? ? ? ? ? .childOption(NioChannelOption.SO_KEEPALIVE, Boolean.TRUE) ? ? ? ? ? ? ? ? ? ?//默認不使用堆外內(nèi)存 ? ? ? ? ? ? ? ? ? .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT) ? ? ? ? ? ? ? ? ? ?//false 不使用堆外內(nèi)存 ? ? ? ? ? ? ? ? ? .childOption(ChannelOption.ALLOCATOR, new UnpooledByteBufAllocator(false)) ? ? ? ? ? ? ? ? ? ?// ? .handler(new LoggingHandler(LogLevel.INFO)) ? ? ? ? ? ? ? ? ? .childHandler(new ChannelInitializer<SocketChannel>() { ? ? ? ? ? ? ? ? ? ? ? ?@Override ? ? ? ? ? ? ? ? ? ? ? ?public void initChannel(SocketChannel ch) throws Exception { ? ? ? ? ? ? ? ? ? ? ? ? ? ?ChannelPipeline p = ch.pipeline(); ? ? ? ? ? ? ? ? ? ? ? ? ? ?// p.addLast(new LoggingHandler(LogLevel.INFO)); ? ? ? ? ? ? ? ? ? ? ? ? ? ?ByteBuf delimiter = Unpooled.copiedBuffer("&".getBytes()); ? ? ? ? ? ? ? ? ? ? ? ? ? ?ch.pipeline() ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?//最大長度 超出最大長度是否立即拋出異常 是否除去分隔符 特殊分隔符 ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? .addLast(new DelimiterBasedFrameDecoder(10, true, true, delimiter)); ? ? ? ? ? ? ? ? ? ? ? ? ? ?ch.pipeline() ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? .addLast(new EchoServerHandler()); ? ? ? ? ? ? ? ? ? ? ? } ? ? ? ? ? ? ? ? ? }); ? ? ? ? ? ? ?ChannelFuture f = b.bind(8090).sync(); ? ? ? ? ? ?f.channel().closeFuture().sync(); ? ? ? } finally { ? ? ? ? ? ?workerGroup.shutdownGracefully(); ? ? ? } ? } } ? @ChannelHandler.Sharable class EchoServerHandler extends ChannelInboundHandlerAdapter { ? ?@Override ? ?public void channelRead(ChannelHandlerContext ctx, Object msg) { ? ? ? ?System.out.println("Receive client : [" + ((ByteBuf) msg).toString(CharsetUtil.UTF_8) + "]"); ? } }

package io.netty.example.decode; ? import io.netty.bootstrap.Bootstrap; import io.netty.channel.*; import io.netty.channel.nio.NioEventLoopGroup; import io.netty.channel.socket.SocketChannel; import io.netty.channel.socket.nio.NioSocketChannel; import io.netty.handler.logging.LogLevel; import io.netty.handler.logging.LoggingHandler; ? ? public final class EchoClient { ? ? ?public static void main(String[] args) throws Exception { ? ? ? ?// Configure the client. ? ? ? ?EventLoopGroup group = new NioEventLoopGroup(); ? ? ? ?try { ? ? ? ? ? ?Bootstrap b = new Bootstrap(); ? ? ? ? ? ?b.group(group) ? ? ? ? ? ? .channel(NioSocketChannel.class) ? ? ? ? ? ? .option(ChannelOption.TCP_NODELAY, true) ? ? ? ? ? ? .handler(new ChannelInitializer<SocketChannel>() { ? ? ? ? ? ? ? ? @Override ? ? ? ? ? ? ? ? public void initChannel(SocketChannel ch) throws Exception { ? ? ? ? ? ? ? ? ? ? ChannelPipeline p = ch.pipeline(); ? ? ? ? ? ? ? ? ? // p.addLast(new LoggingHandler(LogLevel.INFO)); ? ? ? ? ? ? ? ? ? ? p.addLast(new EchoClientHandler()); ? ? ? ? ? ? ? ? } ? ? ? ? ? ? }); ? ? ? ? ? ? ?// Start the client. ? ? ? ? ? ?ChannelFuture f = b.connect("127.0.0.1", 8090).sync(); ? ? ? ? ? ? ?// Wait until the connection is closed. ? ? ? ? ? ?f.channel().closeFuture().sync(); ? ? ? } finally { ? ? ? ? ? ?// Shut down the event loop to terminate all threads. ? ? ? ? ? ?group.shutdownGracefully(); ? ? ? } ? } } ?

package io.netty.example.decode; ? import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import io.netty.channel.ChannelHandlerContext; import io.netty.channel.ChannelInboundHandlerAdapter; ? import java.util.concurrent.TimeUnit; ? /** * Handler implementation for the echo client. It initiates the ping-pong * traffic between the echo client and server by sending the first message to * the server. */ public class EchoClientHandler extends ChannelInboundHandlerAdapter { ? ? ?private final ByteBuf firstMessage; ? ? ?/** ? ? * Creates a client-side handler. ? ? */ ? ?//TODO 修改1234567890 看看10位數(shù)字 和 非10位數(shù)字的區(qū)別 ? ?public EchoClientHandler() { ? ? ? ?firstMessage = Unpooled.wrappedBuffer("1234567890".getBytes()); ? } ? ? ?@Override ? ?public void channelActive(ChannelHandlerContext ctx) { ? ? ? ?System.out.println("客戶端發(fā)送消息" + firstMessage.toString()); ? ? ? ? ?ctx.writeAndFlush(firstMessage); ? ? } ? ? ?@Override ? ?public void channelRead(ChannelHandlerContext ctx, Object msg) { ? ? ? // ctx.write(msg); ? } ? ? ?@Override ? ?public void channelReadComplete(ChannelHandlerContext ctx) throws InterruptedException { ? ? ? ?TimeUnit.SECONDS.sleep(3); ? ? ? ?ctx.flush(); ? } ? ? ?@Override ? ?public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { ? ? ? // cause.printStackTrace(); ? ? ? ?ctx.close(); ? } } ?

我們依然通過 telnet 模擬客戶端發(fā)送數(shù)據(jù)，觀察代碼輸出的結(jié)果，可以發(fā)現(xiàn)由于 maxLength 設(shè)置的只有 10，所以在解析到第三個消息時拋出異常。

客戶端輸入：

java telnet localhost 8088 ? Trying ::1... ? Connected to localhost. ? Escape character is '^]'. ? hello&world&1234567890ab ?

服務(wù)端輸出：

java Receive client : [hello] Receive client : [world] 九月 25, 2020 8:46:01 下午 io.netty.channel.DefaultChannelPipeline onUnhandledInboundException 警告: An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception. io.netty.handler.codec.TooLongFrameException: frame length exceeds 10: 13 - discarded at io.netty.handler.codec.DelimiterBasedFrameDecoder.fail(DelimiterBasedFrameDecoder.java:302) at io.netty.handler.codec.DelimiterBasedFrameDecoder.decode(DelimiterBasedFrameDecoder.java:268) at io.netty.handler.codec.DelimiterBasedFrameDecoder.decode(DelimiterBasedFrameDecoder.java:218)

長度域:LengthFieldBasedFrameDecoder

長度域解碼器 LengthFieldBasedFrameDecoder 是解決 TCP 拆包/粘包問題最常用的解碼器。 它基本上可以覆蓋大部分基于長度拆包場景，開源消息中間件 RocketMQ 就是使用 LengthFieldBasedFrameDecoder 進行解碼的。LengthFieldBasedFrameDecoder 相比 FixedLengthFrameDecoder 和 DelimiterBasedFrameDecoder 要復(fù)雜一些，接下來我們就一起學習下這個強大的解碼器。

首先我們同樣先了解 LengthFieldBasedFrameDecoder 中的幾個重要屬性，這里我主要把它們分為兩個部分：長度域解碼器特有屬性以及與其他解碼器(如特定分隔符解碼器)的相似的屬性。

• 長度域解碼器特有屬性

java // 長度字段的偏移量，也就是存放長度字段的位置 // 如 長度字段是0 那么長度字段放在了最前面 即 數(shù)據(jù)包的起始位置 private final int lengthFieldOffset; ? // 長度字段所占用的字節(jié)數(shù) // 即長度字段占用數(shù)據(jù)包的字節(jié)數(shù) private final int lengthFieldLength; ? /* * 消息長度的修正值,即根據(jù) legnth + lengthAdjustment = 內(nèi)容真正的長度 * * 在很多較為復(fù)雜一些的協(xié)議設(shè)計中，長度域不僅僅包含消息的長度，而且包含其他的數(shù)據(jù)，如版本號、數(shù)據(jù)類型、數(shù)據(jù)狀態(tài)等，那么這時候我們需要使用 lengthAdjustment 進行修正。假如長度域的值為14, 其中content的長度是12 那lengthAdjustment = 12 - 14 =-2 * * lengthAdjustment = 解碼前消息內(nèi)容字段的起始位置 - 長度字段偏移量 -長度字段所占用的字節(jié)數(shù) * lengthAdjustment = initialBytesToStrip - lengthFieldOffset -lengthFieldLength */ ? private final int lengthAdjustment; ? ? // 表示解碼后需要跳過的初始字節(jié)數(shù)，也就是消息內(nèi)容字段的起始位置。 private final int initialBytesToStrip; ? // 長度字段結(jié)束的偏移量，lengthFieldEndOffset = lengthFieldOffset + lengthFieldLength private final int lengthFieldEndOffset; ?

與固定長度解碼器和特定分隔符解碼器相似的屬性

java private final int maxFrameLength; // 報文最大限制長度 private final boolean failFast; // 是否立即拋出 TooLongFrameException，與 maxFrameLength 搭配使用 private boolean discardingTooLongFrame; // 是否處于丟棄模式 private long tooLongFrameLength; // 需要丟棄的字節(jié)數(shù) private long bytesToDiscard; // 累計丟棄的字節(jié)數(shù) ?

下面我們結(jié)合具體的示例來解釋下每種參數(shù)的組合，其實在 Netty LengthFieldBasedFrameDecoder 源碼的注釋中已經(jīng)描述得非常詳細，一共給出了 7 個場景示例，理解了這些示例基本上可以真正掌握 LengthFieldBasedFrameDecoder 的參數(shù)用法。

示例 1：典型的基于消息長度 + 消息內(nèi)容的解碼。

java BEFORE DECODE (14 bytes) ? ? ? ? AFTER DECODE (14 bytes) +--------+----------------+ ? ? ?+--------+----------------+ | Length | Actual Content |----->| Length | Actual Content | | 12 ? ? | "HELLO, WORLD" | ? ? ?| 0x000C | "HELLO, WORLD" | +--------+----------------+ ? ? ?+--------+----------------+ ? 上述協(xié)議是最基本的格式，報文只包含消息長度 Length 和消息內(nèi)容 Content 字段，其中 Length 為 16 進制表示，共占用 2 字節(jié)，Length 的值 0x000C 代表 Content 占用 12 字節(jié)。即要傳輸內(nèi)容的字節(jié)數(shù)是12,該協(xié)議對應(yīng)的解碼器參數(shù)組合如下： ? lengthFieldOffset = 0，存放長度數(shù)據(jù)的起始位置, Length 字段就在報文的開始位置即0 ? lengthFieldLength = 2，長度字段所占用的字節(jié)數(shù),協(xié)議設(shè)計的固定長度為2。 ? lengthAdjustment = 0 Length 字段只包含內(nèi)容長度，不需要做任何修正。 ? initialBytesToStrip = 0 表示解碼后需要跳過的初始字節(jié)數(shù)，也就是消息內(nèi)容字段的起始位置。解碼后14字節(jié)。表示內(nèi)容就是12字節(jié),我們可以先讀2字節(jié)長度。再根據(jù)長度讀取12字節(jié)的真正內(nèi)容。

示例 2：解碼結(jié)果需要截斷。

java BEFORE DECODE (14 bytes) ? ? ? ? AFTER DECODE (12 bytes) +--------+----------------+ ? ? ?+----------------+ | Length | Actual Content |----->| Actual Content | | 12 ? ? | "HELLO, WORLD" | ? ? ?| "HELLO, WORLD" | +--------+----------------+ ? ? ?+----------------+ ? 示例 2 和示例 1 的區(qū)別在于解碼后的結(jié)果只包含消息內(nèi)容，其他的部分是不變的。其中 Length 為 16 進制表示，共占用 2 字節(jié)，Length 的值 0x000C 代表 Content 占用 12 字節(jié)。該協(xié)議對應(yīng)的解碼器參數(shù)組合如下： ? - lengthFieldOffset = 0，存放長度數(shù)據(jù)的起始位置,因為 Length 字段就在報文的開始位置。 - lengthFieldLength = 2，長度字段所占用的字節(jié)數(shù),協(xié)議設(shè)計的固定長度。 - lengthAdjustment = 0 Length 字段只包含消息長度，不需要做任何修正。 - initialBytesToStrip = 2 表示需要跳過2字節(jié)才是真正的消息內(nèi)容

示例 3：長度字段包含消息長度和消息內(nèi)容所占的字節(jié)。

java BEFORE DECODE (14 bytes) ? ? ? ? AFTER DECODE (14 bytes) +--------+----------------+ ? ? ?+--------+----------------+ | Length | Actual Content |----->| Length | Actual Content | | 14 ? ? | "HELLO, WORLD" | ? ? ?| 0x000E | "HELLO, WORLD" | +--------+----------------+ ? ? ?+--------+----------------+ 與前兩個示例不同的是，示例 3 的 Length 字段包含 Length 字段自身的固定長度以及 Content 字段所占用的字節(jié)數(shù)，Length 的值為 0x000E(2 + 12 = 14 字節(jié))，在 Length 字段值(14 字節(jié))的基礎(chǔ)上做 lengthAdjustment(-2)的修正，才能得到真實的 Content 字段長度，所以對應(yīng)的解碼器參數(shù)組合如下： ? ? lengthFieldOffset = 0，因為 Length 字段就在報文的開始位置。 lengthFieldLength = 2，協(xié)議設(shè)計的固定長度。 lengthAdjustment = -2，Actual Content是12,長度字段值為14,14 + (-2) =12 ，即需要減 2 才是拆包所需要的長度。 ? initialBytesToStrip = 0，解碼后內(nèi)容依然是 Length + Content，不需要跳過任何初始字節(jié)。解碼前總長度14-解碼后總長度14=0

示例 4：基于長度字段偏移的解碼。

java BEFORE DECODE (17 bytes) ? ? ? ? ? ? ? ? ? ? ?AFTER DECODE (17 bytes) +----------+----------+----------------+ ? ? ?+----------+----------+------------- | Header| ?Length| Actual Content |-----> ? ?|Header | ?Length ?| Actual Content| | ?2 | 12 | "HELLO, WORLD" | ? ? ? ? ?|0xCAFE | 0x00000C | "HELLO, WORLD"| +----------+----------+----------------+ ? ? ?+----------+----------+------------- ? 示例 4 中 Header 2字節(jié), Length3字節(jié),Length 字段不再是報文的起始位置,Length 字段的值為 0x00000C，表示 Content 字段占用 12 字節(jié)，該協(xié)議對應(yīng)的解碼器參數(shù)組合如下： ? - lengthFieldOffset = 2，需要跳過 Header 1 所占用的 2 字節(jié)，才是 Length 的起始位置。 - lengthFieldLength = 3，協(xié)議設(shè)計的固定長度。 - lengthAdjustment = 0，Length 字段只包含消息長度，不需要做任何修正。before和after一樣 - initialBytesToStrip = 0，解碼后內(nèi)容依然是完整的報文，不需要跳過任何初始字節(jié)。

示例 5：長度字段與內(nèi)容字段不再相鄰。

java BEFORE DECODE (17 bytes) ? ? ? ? ? ? ? ? ? ? ?AFTER DECODE (17 bytes) +----------+----------+----------------+ ? ? ?+----------+----------+------------- | ?Length| Header| Actual Content |-----> |Length ?| Header | Actual Content | | ?12 | ?2 | "HELLO, WORLD" | ? ? |0x00000C| 0xCAFE | "HELLO, WORLD" | +----------+----------+----------------+ ? ? ?+----------+----------+------------- 示例 5 中的 Length 字段之后是 Header，Length 與 Content 字段不再相鄰。Length 字段所表示的內(nèi)容略過了 Header 1 字段，所以也需要通過 lengthAdjustment 修正才能得到 Header + Content 的內(nèi)容。示例 5 所對應(yīng)的解碼器參數(shù)組合如下： ? lengthFieldOffset = 0，因為 Length 字段就在報文的開始位置。 ? lengthFieldLength = 3，協(xié)議設(shè)計的固定長度。 ? lengthAdjustment = 2 ? initialBytesToStrip = 0，解碼后內(nèi)容依然是完整的報文，不需要跳過任何初始字節(jié)。 解碼前總長度17-解碼后總長度17

示例 6：基于長度偏移和長度修正的解碼。

java BEFORE DECODE (16 bytes) ? ? ? ? ? ? ? ? ? ? ? AFTER DECODE (13 bytes) +------+--------+------+----------------+ ? ? ?+------+----------------+ | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content | | 1 ? ?| 12 ? ? | 1 ? ?| "HELLO, WORLD" | ? ? ?| 0xFE | "HELLO, WORLD" | +------+--------+------+----------------+ ? ? ?+------+----------------+ 示例 6 中 Length 字段前后分為別 HDR1 和 HDR2 字段，各占用 1 字節(jié)，所以既需要做長度字段的偏移，也需要做 lengthAdjustment 修正，具體修正的過程與 示例 5 類似。對應(yīng)的解碼器參數(shù)組合如下： ? - lengthFieldOffset = 1，需要跳過 HDR1 所占用的 1 字節(jié)，才是 Length 的起始位置。 - lengthFieldLength = 2，協(xié)議設(shè)計的固定長度。 - lengthAdjustment = 1，length: 12 + HDR2:1 =13 真正的內(nèi)容長度13 - initialBytesToStrip = 3，解碼后跳過 HDR1 和 Length 字段，共占用 3 字節(jié)。 解碼前總長度16-解碼后總長度13=3

示例 7：長度字段包含除 Content 外的多個其他字段。

java BEFORE DECODE (16 bytes) ? ? ? ? ? ? ? ? ? ? ? AFTER DECODE (13 bytes) +------+--------+------+----------------+ ? ? ?+------+----------------+ | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content | | 1 ? ?| 16 ? ? | 3 ? ?| "HELLO, WORLD" | ? ? ?| 0xFE | "HELLO, WORLD" | +------+--------+------+----------------+ ? ? ?+------+----------------+ 示例 7 與 示例 6 的區(qū)別在于 Length 字段記錄了整個報文的長度，包含 Length 自身所占字節(jié)2、HDR1占用字節(jié)1 、HDR2占用字節(jié)1 以及 Content 字段的長度12，解碼器需要知道如何進行 lengthAdjustment 調(diào)整，才能得到 HDR2 和 Content 的內(nèi)容。所以我們可以采用如下的解碼器參數(shù)組合： ? - lengthFieldOffset = 1，需要跳過 HDR1 所占用的 1 字節(jié)，才是 Length 的起始位置。 - lengthFieldLength = 2，協(xié)議設(shè)計的固定長度。 - lengthAdjustment = -3，Actual Content 13 - 長度域的值16=-3 initialBytesToStrip = 3，解碼前16 - ?解碼后 13 = 3 ?

以上 7 種示例涵蓋了 LengthFieldBasedFrameDecoder 大部分的使用場景，你是否學會了呢？最后留一個小任務(wù)，在上一節(jié)課程中我們設(shè)計了一個較為通用的協(xié)議，如下所示。如何使用長度域解碼器 LengthFieldBasedFrameDecoder 完成該協(xié)議的解碼呢？抓緊自己嘗試下吧。

+---------------------------------------------------------------+ | 魔數(shù) 2byte | 協(xié)議版本號 1byte | 序列化算法 1byte | 報文類型 1byte ?| +---------------------------------------------------------------+ | 狀態(tài) 1byte | ? ? ? ?保留字段 4byte ? ? | ? ? ?數(shù)據(jù)長度 4byte ? ? | +---------------------------------------------------------------+ | ? ? ? ? ? ? ? ? ? 數(shù)據(jù)內(nèi)容 (長度不定) ? ? ? ? ? ? ? ? ? ? ? ? ?| +---------------------------------------------------------------+

編碼需要使用LengthFieldPrepender

總結(jié)

本節(jié)課我們介紹了三種常用的解碼器，從中我們可以體會到 Netty 在設(shè)計上的優(yōu)雅，只需要調(diào)整參數(shù)就可以輕松實現(xiàn)各種功能。在健壯性上，Netty 也考慮得非常全面，很多邊界情況 Netty 都貼心地增加了保護性措施。實現(xiàn)一個健壯的解碼器并不容易，很可能因為一次解析錯誤就會導致解碼器一直處理錯亂的狀態(tài)。如果你使用了基于長度編碼的二進制協(xié)議，那么推薦你使用 LengthFieldBasedFrameDecoder，它已經(jīng)可以滿足實際項目中的大部分場景，基本不需要再自定義實現(xiàn)了。希望朋友們在項目開發(fā)中能夠?qū)W以致用。文章來源地址http://www.zghlxwxcb.cn/news/detail-639239.html