5.3 Netty 编码、解码框架

转载自：http://www.tianshouzhi.com/api/tutorials/netty/345

 2017-02-09 21:34:39  2,822  0

Netty提供了大量的网络协议数据格式的编码解码器，从而简化开发者的使用。例如你想开发一个基于Netty的邮件服务器，你将会发现Netty针对POP3、IMAP、SMTP协议的数据格式都提供了相应的编码解码器。

本节我们将从整体结构上对Netty的编码解码技术框架进行介绍。

解码器Decoder

Netty中decoder主要分成2类：

1、将字节流转换成某种协议数据格式：ByteToMessageDecoder和ReplayingDecoder

2、将一种协议数据格式转换成另一种数据格式：MessageToMessageDecoder

因为decoder是负责将输入(inbound)的数据转换成特定协议的数据格式，因此，所有的decoder都实现了ChannelInboundHandler接口。

抽象类ByteToMessageDecoder

将二进制数据转换成程序使用的数据格式是最常见的一个任务，Netty为此提供了一个抽象基类ByteToMessageDecoder。

从名称上来看ByteToMessageDecoder的话，Byte指的是当然就是接受到的二进制数据，而Message指的是根据protocol规定将二进制数据解析出来的有效信息，例如一个请求或者响应。有效信息Message只是一个概念上称呼，通常需要一个载体，以便之后可以方便的操作其包含的数据，显然在java中，Message的载体应该是一个java对象。例如我们会将Http请求解析成一个HttpServletRequest对象。

将byte转换成message最大的难点在于tcp的拆包 ，拆包意味着我们接受的二进制数据还不足以构造成一个完整的message。如果遇到这种情况，应该怎么处理呢？

第一种方案：线程一直等待，直到byte可以构造成一个完整的message。这种方式最大的问题在于，如过剩余部分数据发送很慢，线程一直被占用，不能去处理其他客户端的请求。

第二种方案：为每个客户端连接(SocketChannel)定义一个应用层面的缓存。线程处理某个SocketChannel时，如果遇到byte不足以构成一个message，则将byte到放入缓存中，线程继续去处理其他的SocketChannel的任务。

显然ByteToMessageDecoder采用的是第二种方案，在ByteToMessageDecoder 中，有一个ByteBuf类型的字段，这就是上面所说的缓存。如果每个SocketChannel实例都在其ChannelPipeline中添加了一个ByteToMessageDecoder类型实例，那么就相当于每个SocketChannel都有了自己关联的缓存。

特别要注意的是：ByteToMessageDecoder实现类上是不允许添加@Sharable注解的，添加@Sharable表示多个SocketChannel可以共用一个ChannelHandler实例，但是明显ByteToMessageDecoder是特殊的，如果多个SocketChannel共用一个ByteToMessageDecoder实例，会造成缓存中的数据分不清到底是哪个SocketChannel的。

以下是ByteToMessageDecoder的相关源码：

1. public abstract class ByteToMessageDecoder extends ChannelInboundHandlerAdapter {
2.  
3. ByteBuf cumulation;//cumulation用户缓存解析后的二进制数据
4. private boolean singleDecode;//在解析数据时，是否每次只回调一次decode方法，后面介绍
5. private boolean decodeWasNull;
6. private boolean first;//是否是第一次接受到输入的数据
7. ....
8. //当接受到数据时，channelRead方法会被回调
9. //关于参数Object msg的说明：由于ByteToMessageDecoder只处理二进制数据 ，因此Object类型应该是ByteBuf。
10. //如果不是，将会直接交给pipeline中下一个handler处理。
11.     @Override
12.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
13.         if (msg instanceof ByteBuf) {//如果msg类型是ByteBuf，进行解析
14.         //out用于存储l解析二进制流得到的结果，一个二进制流可能会解析出多个消息，所以out是一个list
15.             RecyclableArrayList out = RecyclableArrayList.newInstance();
16.             try {
17.                 ByteBuf data = (ByteBuf) msg;//将msg强转为ByteBuf类型
18.                 //判断cumulation == null;并将结果赋值给first。因此如果first为true，则表示第一次接受到数据                
19.                 first = cumulation == null;
20.                 if (first) {//如果是第一次接受到数据，直接将接受到的数据赋值给缓存对象cumulation
21.                     cumulation = data;
22.                 } else {//如果不是第一次接受到数据
23.                     if (cumulation.writerIndex() > cumulation.maxCapacity() - data.readableBytes()
24.                             || cumulation.refCnt() > 1) {
25.                             //如果cumulation中的剩余空间，不足以存储接收到的data
26.                               expandCumulation(ctx, data.readableBytes());//将cumulation扩容
27.                             }
28.                           cumulation.writeBytes(data);//将data拷贝到cumulation中
29.                           data.release();
30.                     }
31.                    //调用callDecode，开始解析cumulation中的数据，解析结果放到out中，这是一个list
32.                    //因为我们可能根据cumulation中的数据，解析出多个有效数据
33.                 callDecode(ctx, cumulation, out);
34.             } catch (DecoderException e) {
35.                 throw e;
36.             } catch (Throwable t) {
37.                 throw new DecoderException(t);
38.             } finally {
39.                //如果cumulation没有数据可读了，说明所有的二进制数据都被解析过了
40.                //此时对cumulation进行释放，以节省内存空间。
41.                      //反之cumulation还有数据可读，那么if中的语句不会运行，因为不对cumulation进行释放
42.                      //因此也就缓存了用户尚未解析的二进制数据。
43.                 if (cumulation != null && !cumulation.isReadable()) {
44.                        cumulation.release();
45.                               cumulation = null;
46.                 }
47.                 int size = out.size();//获得解析二进制流得到的消息的个数
48.                 decodeWasNull = size == 0;
49.                                 //迭代每一个解析出来的消息，调用下一个ChannelHandler进行处理
50.                 for (int i = 0; i < size; i ++) {
51.                     ctx.fireChannelRead(out.get(i));
52.                 }
53.                 out.recycle();
54.             }
55.         } else {//如果msg类型是不是ByteBuf，直接调用下一个handler进行处理
56.             ctx.fireChannelRead(msg);
57.            }
58.   }
59.  
60.      //callDecode方法主要用于解析cumulation 中的数据，并将解析的结果放入List<Object> out中。
61.      //由于cumulation中缓存的二进制数据，可能包含了出多条有效信息，因此在callDecode方法中，默认会调用多次decode方法
62.      //我们在覆写decode方法时，每次只解析一个消息，添加到out中，callDecode通过多次回调decode
63.      //每次传递进来都是相同的List<Object> out实例，因此每一次解析出来的消息，都存储在同一个out实例中。
64.      //当cumulation没有数据可以继续读，或者某次调用decode方法后，List<Object> out中元素个数没有变化，则停止回调decode方法。
65.    protected void callDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
66.         try {
67.           while (in.isReadable()) {//如果in，即cumulation中有数据可读的话，一直循环调用decode
68.                int outSize = out.size();//获取上一次decode方法调用后，out中元素数量，如果是第一次调用，则为0。
69.               int oldInputLength = in.readableBytes();//上次decode方法调用后，in的剩余可读字节数
70.               //回调decode方法，由开发者覆写，用于解析in中包含的二进制数据，并将解析结果放到out中。
71.               decode(ctx, in, out);
72.                 // See https://github.com/netty/netty/issues/1664
73.                     if (ctx.isRemoved()) {
74.                        break;
75.                    }
76.                 //outSize是上一次decode方法调用时out的大小，out.size()是当前out大小
77.                 //如果二者相等，则说明当前decode方法调用没有解析出有效信息。
78.                 if (outSize == out.size()) {
79.                 //此时，如果发现上次decode方法和本次decode方法调用候，in中的剩余可读字节数相同
80.                 //则说明本次decode方法没有读取任何数据解析
81.                 //(可能是遇到半包等问题，即剩余的二进制数据不足以构成一条消息)，跳出while循环。
82.                  if (oldInputLength == in.readableBytes()) {
83.                    break;
84.                  } else {
85.                    continue;
86.                   }
87.                 }
88.                 //处理人为失误 。如果走到这段代码，则说明outSize != out.size()。
89.                 //也就是本次decode方法实际上是解析出来了有效信息放到out中。
90.                 //但是oldInputLength == in.readableBytes()，说明本次decode方法调用并没有读取任何数据
91.                 //但是out中元素却添加了。
92.                 //这可能是因为开发者错误的编写了代码，例如mock了一个消息放到List中。
93.                 if (oldInputLength == in.readableBytes()) {
94.                     throw new DecoderException(
95.                             StringUtil.simpleClassName(getClass()) +
96.                             ".decode() did not read anything but decoded a message.");
97.                 }
98.  
99.                 if (isSingleDecode()) {
100.                     break;
101.                 }
102.             }
103.         } catch (DecoderException e) {
104.             throw e;
105.         } catch (Throwable cause) {
106.             throw new DecoderException(cause);
107.         }
108.     }
109.  
110.     /**抽象方法，由子类覆盖，建议在decode方法中，一次只解析一条信息，不足以构成一条信息的数据，不要读取*/
111.     protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
112.  
113.     ...
114. }

以Client给Server发送一个请求为例，有可能Server一开始没有接收到完整的信息，因为数据都是以二进制流的方式传输，并不能保证一条消息的二进制数据会一次性被完整的接受。ByteToMessageDecoder可以帮助我们将不完整的二进制数据进行缓存，直到能构成完整的消息时，才开始进行处理。

通过以上分析，ByteToMessageDecoder在使用时，我们只需要覆写其decode方法即可。

protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;

   假设我们要读取一个字节流，这个字节流中仅仅包含一些int型数字，假设协议是这样规定的。那么每个数字都是一条消息，因此需要单独处理。在解码完成后，ByteToMessageDecoder会自动迭代out中的元素，逐一交给pipeline中下一个ChannelInboundHandler进行处理了。假设我们编写了一个ToIntegerDecoder来实现解码，那么整个流程如下图所示。

   参数ByteBuf中包含了需要读取的二进制数据，然后将解析后的请求信息放入List中，如果解析出了多个请求，则每条信息都要放入List中。解析完成之后，则会将List中的消息交给pipeline中后续的handler处理。如果ByteBuf中的字节数不足以构成一条消息，那么只要不解析即可，此时List为空，为空的情况下，后面的handler不会被调用。此时数据还在ByteBuf中，也就是我们前面提到的这个类可以帮助我们缓存不完整的数据的功能。decode方法可能会被回调多次，当ByteBuf中没有更多的数据可以读取，获取某次回调之后List中没有添加新的元素时，回调结束。

ToIntegerDecoder源码如下所示：

1. public class ToIntegerDecoder extends ByteToMessageDecoder {
2.     @Override
3.     protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
4.         if (in.readableBytes() >= 4) {//1个int型数字有4个字节，没有4个字节就说明不足以构成一条请求消息
5.             out.add(in.readInt());
6.         }
7.     }
8. }

尽管ByteToMessageDecoder这种设计方式已经可以帮助我们简化开发，但是我们发现了每次都必须判断ByteBuf中是否有足够的数据可以读取，这有点让人讨厌。下面我们要讨论的ReplayingDecoder可以省略这个步骤，代价是你需要多花一点时间来理解。

如果想获得更加实际的ByteToMessageDecoder的使用案例，可以参考Netty提供的LineBasedFrameDecoder，这是一个按行(\n、\r\n)分隔的解码器。

抽象类ReplayingDecoder

ReplayingDecoder继承自ByteToMessageDecoder，其可以让我们在读取数据时不需要调用ByteBuf的readableBytes()方法。这个功能是通过其自定义的ReplayingDecoderBuffer来完成的。

其类声明如下：

public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder

泛型参数S表示用于状态管理的类型,Void表示不需要执行任何操作。下面是ToIntegerDecoder基于ReplayingDecoder版本的实现

1. public class ToIntegerDecoder2 extends ReplayingDecoder<Void> {
2.     @Override
3.     protected void decode(ChannelHandlerContext ctx,
4.                           ByteBuf in,//注意这里传入的ByteBuf实际上是 ReplayingDecoderBuffer
5.                           List<Object> out) throws Exception {
6.         out.add(in.readInt());
7.     }
8. }

当没有足够的数据可以读时，readInt()将会抛出一个Error错误，这个错误将会被捕获，当有很多的数据可以读时，decode方法将会被再次调用。

此外需要注意： 

1、ReplayingDecoderBuffer并不支持在ByteBuf中定义的所有方法，调用了一个不支持的方法时，将会抛出UnsupportedOperationException 

2、从效率上来说，ReplayingDecoder略低于ByteToMessageDecoder 

如果想获得更多ReplayingDecoder的使用案例，可以参考Netty提供的HttpObjectDecoder，这是一个Http协议的解码器。

抽象类MessageToMessageDecoder

ByteToMessageDecoder主要是将二进制数据流转换成协议规定的数据格式，例如，在J2EE开发中，我们可能会将二进制数据流转换成一个HttpServletRequest对象，从这个对象中我们可以HTTP协议规定一个请求应该包含的所有信息。

而在实际开发过程中，有可能我们并不关心HttpServletRequest中的所有数据，只关心其通过表单提交的参数(请求体)，因此有了Struts、SpringMvc这样的框架，可以将HttpServletRequest中的请求参数封装到我们自己定义的POJO类中。

MessageToMessageDecoder做的就是这样的事，其将通过ByteToMessageDecoder解码的消息，再次进行解码，再交由pipeline之后的handler进行处理。

MessageToMessageDecoder的类声明如下：

public abstract class MessageToMessageDecoder<I> extends ChannelInboundHandlerAdapter 

其中泛型参数I表示我们要解码的消息类型。例如上一节，我们在ToIntegerDecoder中，把二进制字节流转换成了一个int类型的整数。

例如，现在我们想编写一个IntegerToStringDecoder，把前面编写的ToIntegerDecoder输出的int参数转换成字符串，此时泛型I就应该是Integer类型。

类似的，MessageToMessageDecoder也有一个decode方法需要覆盖 ，如下：

/**

参数msg，就是之前经过解码的消息，例如 ByteToMessageDecoder解码后的消息。

List<Object> out参数：将msg经过转换后的到的新的message，放到List<Object> out中

*/

protected abstract void decode(ChannelHandlerContext ctx, I msg, List<Object> out) throws Exception;

IntegerToStringDecoder源码如下所示：

1. public class IntegerToStringDecoder extends
2.         MessageToMessageDecoder<Integer> {
3.     @Override
4.     protected void decode(ChannelHandlerContext ctx, Integer msg, List<Object> out) throws Exception {
5.         out.add(String.valueOf(msg));//将Integer转成String类型，加入到 List<Object> out中
6.     }
7. }

此时我们应该按照如下顺序组织ChannelPipieline中ToIntegerDecoder和IntegerToStringDecoder 的关系：

ChannelPipieline ch=....

ch.addLast(new ToIntegerDecoder());

ch.addLast(new IntegerToStringDecoder());

io.netty.handler.codec.http.HttpObjectAggregator继承了MessageToMessageDecoder，感兴趣的读者可以阅读其源码获得更多如何使用MessageToMessageDecoder的信息。

编码器Encoder

Encoders用于编码。对比Decoders，Encoders都实现了ChannelOutboundHandler接口。Netty提供了：

MessageToByteEncoder：将消息编码成二进制字符串

MessageToMessageEncoder：将一个消息编码成另一个消息

抽象类MessageToByteEncoder

MessageToByteEncoder是一个泛型类，泛型参数I表示将需要编码的对象的类型，编码的结果是将信息转换成二进制流放入ByteBuf中。子类通过覆写其抽象方法encode，来实现编码，如下所示：

public abstract class MessageToByteEncoder<I> extends ChannelOutboundHandlerAdapter {

....

protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;

}

MessageToByteEncoder使用案例：

1. public class IntegerToByteEncoder extends MessageToByteEncoder<Integer> {
2.     @Override
3.     protected void encode(ChannelHandlerContext ctx, Integer msg, ByteBuf out) throws Exception {
4.         out.writeInt(msg);//将Integer转成二进制字节流写入ByteBuf中
5.     }
6. }

类似的MessageToMessageEncoder不再赘述。

编码解码器Codec

编码解码器同时具有编码与解码功能，Netty提供了抽象基类： ByteToMessageCodec 、MessageToMessageCodec。

根据我们的经验，每个codec中应该同时具有encode和decode两个抽象方法要覆盖：

ByteToMessageCodec：

1. public abstract class ByteToMessageCodec<I> extends ChannelDuplexHandler {
2.  
3.     ...
4.     protected abstract void encode(ChannelHandlerContext ctx, I msg, ByteBuf out) throws Exception;
5.  
6.     
7.    protected abstract void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception;
8.  
9.     ...
10. }

MessageToMessageCodec：

1. public abstract class MessageToMessageCodec<INBOUND_IN, OUTBOUND_IN> extends ChannelDuplexHandler {
2.  
3.     ....
4.     protected abstract void encode(ChannelHandlerContext ctx, OUTBOUND_IN msg, List<Object> out)
5.             throws Exception;
6.  
7.  
8.     protected abstract void decode(ChannelHandlerContext ctx, INBOUND_IN msg, List<Object> out)
9.             throws Exception;
10. }