netty解码器详解（小白也能看懂！）

什么是编解码器？

首先，我们回顾一下netty的组件设计：Netty的主要组件有Channel、EventLoop、ChannelFuture、ChannelHandler、ChannelPipe等。

ChannelHandler

ChannelHandler充当了处理入站和出站数据的应用程序逻辑的容器。例如，实现ChannelInboundHandler接口（或ChannelInboundHandlerAdapter），你就可以接收入站事件和数据，这些数据随后会被你的应用程序的业务逻辑处理。当你要给连接的客户端发送响应时，也可以从ChannelInboundHandler冲刷数据。你的业务逻辑通常写在一个或者多个ChannelInboundHandler中。ChannelOutboundHandler原理一样，只不过它是用来处理出站数据的。

ChannelPipeline

ChannelPipeline提供了ChannelHandler链的容器。以客户端应用程序为例，如果事件的运动方向是从客户端到服务端的，那么我们称这些事件为出站的，即客户端发送给服务端的数据会通过pipeline中的一系列ChannelOutboundHandler，并被这些Handler处理，反之则称为入站的。

netty解码器详解（小白也能看懂！）

编码解码器

当你通过Netty发送或者接受一个消息的时候，就将会发生一次数据转换。入站消息会被解码：从字节转换为另一种格式（比如java对象）；如果是出站消息，它会被编码成字节。

Netty提供了一系列实用的编码解码器，他们都实现了ChannelInboundHadnler或者ChannelOutcoundHandler接口。在这些类中，channelRead方法已经被重写了。以入站为例，对于每个从入站Channel读取的消息，这个方法会被调用。随后，它将调用由已知解码器所提供的decode()方法进行解码，并将已经解码的字节转发给ChannelPipeline中的下一个ChannelInboundHandler。

解码器

抽象基类ByteToMessageDecoder

netty解码器详解（小白也能看懂！）

由于你不可能知道远程节点是否会一次性发送一个完整的信息，tcp有可能出现粘包拆包的问题，这个类会对入站数据进行缓冲，直到它准备好被处理。

主要api有两个：

public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {
 protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception;
 protected void decodeLast(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 if (in.isReadable()) {
 // Only call decode() if there is something left in the buffer to decode.
 // See https://github.com/netty/netty/issues/4386
 decodeRemovalReentryProtection(ctx, in, out);
 }
 }
}

decode方法：

必须实现的方法，ByteBuf包含了传入数据，List用来添加解码后的消息。对这个方法的调用将会重复进行，直到确定没有新的元素被添加到该List，或者该ByteBuf中没有更多可读取的字节时为止。然后如果该List不会空，那么它的内容将会被传递给ChannelPipeline中的下一个ChannelInboundHandler。

decodeLast方法：

当Channel的状态变成非活动时，这个方法将会被调用一次。

最简单的例子：

netty解码器详解（小白也能看懂！）

public class ToIntegerDecoder extends ByteToMessageDecoder {
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 if (in.readableBytes() &gt;= 4) {
 out.add(in.readInt());
 }
 }
}

这个例子，每次入站从ByteBuf中读取4字节，将其解码为一个int，然后将它添加到下一个List中。当没有更多元素可以被添加到该List中时，它的内容将会被发送给下一个ChannelInboundHandler。int在被添加到List中时，会被自动装箱为Integer。在调用readInt()方法前必须验证所输入的ByteBuf是否具有足够的数据。

一个实用的例子：

public class MyDecoder extends ByteToMessageDecoder {
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 if (in.readableBytes() &lt; 4) {
 return;
 }
　　　　　//在读取前标记readerIndex
 in.markReaderIndex();
　　　　　//读取头部
 int length = in.readInt();
 if (in.readableBytes() &lt; length) {
　　　　　//消息不完整，无法处理，将readerIndex复位
 in.resetReaderIndex();
 return;
 }
 out.add(in.readBytes(length).toString(CharsetUtil.UTF_8));
 }
}

抽象类ReplayingDecoder

public abstract class ReplayingDecoder&lt;S&gt; extends ByteToMessageDecoder

ReplayingDecoder扩展了ByteToMessageDecoder类，使用这个类，我们不必调用readableBytes()方法。参数S指定了用户状态管理的类型，其中Void代表不需要状态管理。

以上代码可以简化为：

public class MySimpleDecoder extends ReplayingDecoder&lt;Void&gt; {
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 //传入的ByteBuf是ReplayingDecoderByteBuf
 //首先从入站ByteBuf中读取头部，得到消息体长度length，然后读取length个字节， 
 //并添加到解码消息的List中
 out.add(in.readBytes(in.readInt()).toString(CharsetUtil.UTF_8));
 }

如何实现的？

ReplayingDecoder在调用decode方法时，传入的是一个自定义的ByteBuf实现：

final class ReplayingDecoderByteBuf extends ByteBuf

ReplayingDecoderByteBuf在读取数据前，会先检查是否有足够的字节可用，以readInt()为例：

final class ReplayingDecoderByteBuf extends ByteBuf {
 private static final Signal REPLAY = ReplayingDecoder.REPLAY;
 ...... 
 @Override
 public int readInt() {
 checkReadableBytes(4);
 return buffer.readInt();
 }
 private void checkReadableBytes(int readableBytes) {
 if (buffer.readableBytes() &lt; readableBytes) {
 throw REPLAY;
 }
 } 
 ......
}

如果字节数量不够，会抛出一个Error（实际是一个Signal public final class Signal extends Error implements Constant<Signal> ），然后会在上层被捕获并处理，它会把ByteBuf中的ReadIndex恢复到读之前的位置，以供下次读取。当有更多数据可供读取时，该decode()方法将会被再次调用。最终结果和之前一样，从ByteBuf中提取的String将会被添加到List中。

虽然ReplayingDecoder使用方便，但它也有一些局限性：

1. 并不是所有的 ByteBuf 操作都被支持，如果调用了一个不被支持的方法，将会抛出一个 UnsupportedOperationException。

2. ReplayingDecoder 在某些情况下可能稍慢于 ByteToMessageDecoder，例如网络缓慢并且消息格式复杂时，消息被拆成了多个碎片，于是decode()方法会被多次调用反复地解析一个消息。

3. 你需要时刻注意decode()方法在同一个消息上可能被多次调用.。

错误用法：

一个简单的echo服务，客户端在连接建立时，向服务端发送消息（两个1）。服务端需要一次拿到两个Integer，并做处理。

EchoServerHandler

public class EchoServerHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("msg from client: " + msg);
 }
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 cause.printStackTrace();
 ctx.close();
 }
}

EchoClientHandler

public class EchoClientHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 System.out.println("sent to server: 11");
 ctx.writeAndFlush(1);
 
 Thread.sleep(1000);
 ctx.writeAndFlush(1);
 }
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 }
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 cause.printStackTrace();
 ctx.close();
 }
}

解码器

public class MyReplayingDecoder extends ReplayingDecoder&lt;Void&gt; {
 private final Queue&lt;Integer&gt; values = new LinkedList&lt;&gt;();
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 values.add(in.readInt());
 values.add(in.readInt());
 assert values.size() == 2;
 out.add(values.poll() + values.poll());
 }
}

运行程序，就会发现断言失败。

我们通过在decode()方法中打印日志或者打断点的方式，可以看到，decode()方法是被调用了两次的，分别在服务端两次接受到消息的时候：

第一次调用时，由于缓冲区中只有四个字节，在第二句 values.add(in.readInt()) 中抛出了异常REPLAY，在ReplayingDecoder中被捕获，并复位ReadIndex。此时values.size() = 1。

第二次调用时，从头开始读取到两个Integer并放入values，因此values.size() = 3。

正确用法：

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 //清空队列
 values.clear();
 values.add(in.readInt());
 values.add(in.readInt());
 assert values.size() == 2;
 out.add(values.poll() + values.poll());
}

如何提高ReplayingDecoder的性能？如上所说，使用ReplayingDecoder存在对一个消息多次重复解码的问题，我们可以通过Netty提供的状态控制来解决这个问题。

首先我们将消息结构设计为：header（4个字节，存放消息体长度），body（消息体）

根据消息的结构，我们定义两个状态：

public enum MyDecoderState {
 /**
 * 未读头部
 */
 READ_LENGTH,
 /**
 * 未读内容
 */
 READ_CONTENT;
}

EchoClientHandler

public class EchoClientHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelActive(ChannelHandlerContext ctx) throws Exception {
 for (int i = 0; i &lt; 10; i++) {
 System.out.println("sent to server: msg" + i);
 ctx.writeAndFlush("msg" + i);
 }
 }
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 }
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 cause.printStackTrace();
 ctx.close();
 }
}

EchoServerHandler

public class EchoServerHandler extends ChannelInboundHandlerAdapter {
 @Override
 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
 System.out.println("msg from client: " +
 ((ByteBuf) msg).toString(CharsetUtil.UTF_8));
 }
 @Override
 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
 }
 @Override
 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
 cause.printStackTrace();
 ctx.close();
 }
}

解码器

public class IntegerHeaderFrameDecoder extends ReplayingDecoder&lt;MyDecoderState&gt; {
 private int length;
 public IntegerHeaderFrameDecoder() {
 // Set the initial state.
 super(MyDecoderState.READ_LENGTH);
 }
 @Override
 protected void decode(ChannelHandlerContext ctx, ByteBuf in, List&lt;Object&gt; out) throws Exception {
 switch (state()) {
 case READ_LENGTH:
 length = in.readInt();
 checkpoint(MyDecoderState.READ_CONTENT);
 case READ_CONTENT:
 ByteBuf frame = in.readBytes(length);
 checkpoint(MyDecoderState.READ_LENGTH);
 out.add(frame);
 break;
 default:
 throw new Error("Shouldn't reach here.");
 }
 }
}

编码器

public class MyEncoder extends MessageToByteEncoder&lt;String&gt; {
 @Override
 protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
 byte[] b = msg.getBytes();
 int length = b.length;
 //write length of msg
 out.writeInt(length);
 //write msg
 out.writeBytes(b);
 }
}

当头部被成功读取到时，我们调用 checkpoint(MyDecoderState.READ_CONTENT) 设置状态为“未读消息”，相当于设置一个标志位，如果在后续读取时抛出异常，那么readIndex会被复位到上一次你调用checkpoint()方法的地方。下一次接收到消息，再次调用decode()方法时，就能够从checkpoint处开始读取，避免了又从头开始读。

更多解码器：

LineBasedFrameDecoder

这个类在Netty内部也有使用，它使用行尾控制字符（或者）作为分隔符来解析数据。

DelimiterBasedFrameDecoder

使用自定义的特殊字符作为消息的分隔符。

HttpObjectDecoder

一个HTTP数据的解码器。

netty解码器详解（小白也能看懂！）

相关推荐