netty 系列之: 自定义编码解码器

简介在之前的 netty 系列文章中，我们讲到了如何将对象或者 String 转换成为 ByteBuf，通过使用 netty 自带的 encoder 和 decoder 可以实现非常方便的对象和 ByteBuf 之间的转换，然后就可以向 channel 中随意写入对象和字符串了。

使用 netty 自带的编码器当然很好，但是如果你有些特殊的需求，比如希望在编码的过程中对数据进行变换，或者对对象的字段进行选择，那么可能就需要自定义编码解码器了。

自定义编码器自定义编码器需要继承 MessageToByteEncoder 类，并实现 encode 方法，在该方法中写入具体的编码逻辑。

本例我们希望计算 2 的 N 次方，据说将一张纸折叠 100 次可以达到地球到月亮的高度，这么大的数据普通的 number 肯定是装不下的，我们将会使用 BigInteger 来对这个巨大的数字进行保存。

那么对于被编码器来说，则需要将这个 BigInteger 转换成为 byte 数组。同时在 byte 数组读取的过程中，我们需要界定到底哪些 byte 数据是属于同一个 BigInteger 的，这就需要对写入的数据格式做一个约定。

这里我们使用三部分的数据结构来表示一个 BigInteger。第一部分是一个 magic word 也就是魔法词，这里我们使用魔法词“N”，当读取到这个魔法词就表示接下来的数字是 BigInteger。第二部分是表示 bigInteger 数字的 byte 数组的长度，获取到这个长度值，就可以读取到所有的 byte 数组值，最后将其转换成为 BigInteger。

因为 BigInteger 是 Number 的子类，为了更加泛化编码器，我们使用 Number 作为 MessageToByteEncoder 的泛型，核心编码代码如下：

protected void encode(ChannelHandlerContext ctx, Number msg, ByteBuf out) {// 将 number 编码成为 ByteBufBigInteger v;if (msg instanceof BigInteger) {v = (BigInteger) msg;} else {v = new BigInteger(String.valueOf(msg));}

    // 将BigInteger转换成为byte[]数组    byte[] data = v.toByteArray();    int dataLength = data.length;
    // 将Number进行编码    out.writeByte((byte) 'N'); // 魔法词    out.writeInt(dataLength);  // 数组长度    out.writeBytes(data);      // 最终的数据}

自定义解码器有了编码之后的 byte 数组，就可以在解码器中对其解码了。

上一节介绍了，编码过后的数据格式是魔法词 N+数组长度+真正的数据。

其中魔法词长度是一个字节，数组长度是四个字节，前面部分总共是 5 个字节。所以在解码的时候，首先判断 ByteBuf 中可读字节的长度是否小于 5，如果小于 5 说明数据是无效的，可以直接 return。

如果可读字节的长度大于 5，则表示数据是有效的，可以进行数据的解码了。

解码过程中需要注意的是，并不是所有的数据都是我们所希望的格式，如果在读取的过程中读到了我们不认识的格式，那么说明这个数据并不是我们想要的，则可以交由其他的 handler 进行处理。

但是对于 ByteBuf 来说，一旦调用 read 方法，就会导致 reader index 移动位置，所以在真正的读取数据之前需要调用 ByteBuf 的 markReaderIndex 方法，对 readerIndex 进行记录。然后分别读取魔法词、数组长度和剩余的数据，最后将数据转换成为 BigInteger，如下所示：

protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {// 保证魔法词和数组长度有效 if (in.readableBytes() < 5) {return;}in.markReaderIndex();// 检查魔法词 int magicNumber = in.readUnsignedByte();if (magicNumber != 'N') {in.resetReaderIndex();throw new CorruptedFrameException("无效的魔法词: " + magicNumber);}// 读取所有的数据 int dataLength = in.readInt();if (in.readableBytes() < dataLength) {in.resetReaderIndex();return;}// 将剩下的数据转换成为 BigIntegerbyte[] decoded = new byte[dataLength];in.readBytes(decoded);out.add(new BigInteger(decoded));}添加编码解码器到 pipeline 有了两个编码解码器，还需要将其添加到 pipeline 中进行调用。

在实现 ChannelInitializer 中的 initChannel 中，可以对 ChannelPipeline 进行初始化，本例中的初始化代码如下：

// 对流进行压缩 pipeline.addLast(ZlibCodecFactory.newZlibEncoder(ZlibWrapper.GZIP));pipeline.addLast(ZlibCodecFactory.newZlibDecoder(ZlibWrapper.GZIP));

    // 添加number编码解码器    pipeline.addLast(new NumberDecoder());    pipeline.addLast(new NumberEncoder());
    // 添加业务处理逻辑    pipeline.addLast(new CustomProtocolServerHandler());

其中最后一行是真正的业务处理逻辑，NumberDecoder 和 NumberEncoder 是编码和解码器。这里我们还使用了一个 ZlibEncoder 用于对流数据进行压缩，这里使用的压缩方式是 GZIP。

压缩的好处就是可以减少数据传输的数量，提升传输效率。其本质也是一个编码解码器。

计算 2 的 N 次方计算 2 的 N 次方的逻辑是这样的，首先客户端发送 2 给服务器端，服务器端接收到该消息和结果 1 相乘，并将结果写回给客户端，客户端收到消息之后再发送 2 给服务器端，服务器端将上次的计算结果乘以 2，再发送给客户端，以此类推直到执行 N 次。

首先看下客户端的发送逻辑：

// 最大计算 2 的 1000 次方 ChannelFuture future = null;for (int i = 0; i < 1000 && next <= CustomProtocolClient.COUNT; i++) {future = ctx.write(2);next++;}当 next 小于等于要计算的 COUNT 时，就将 2 写入到 channel 中。

对于服务器来说，在 channelRead0 方法中，读取消息，并将其和结果相乘，再把结果写回给客户端。

public void channelRead0(ChannelHandlerContext ctx, BigInteger msg) throws Exception {    // 将接收到的msg乘以2，然后返回给客户端    count++;    result = result.multiply(msg);    ctx.writeAndFlush(result);}

客户端统计读取到的消息个数，如果消息个数=COUNT，说明计算完毕，就可以将结果保存起来供后续使用，其核心代码如下：

public void channelRead0(ChannelHandlerContext ctx, final BigInteger msg) {    receivedMessages ++;    if (receivedMessages == CustomProtocolClient.COUNT) {        // 计算完毕，将结果放入answer中        ctx.channel().close().addListener(future -> {            boolean offered = answer.offer(msg);            assert offered;        });    }}

总结本文实现了一个 Number 的编码解码器，事实上你可以自定义实现任何对象的编码解码器。

本文的例子可以参考：learn-netty4

本文已收录于 http://www.flydean.com/13-netty-customprotocol/

最通俗的解读，最深刻的干货，最简洁的教程，众多你不知道的小技巧等你来发现！

欢迎关注我的公众号:「程序那些事」,懂技术，更懂你！