理解netty對protocol buffers的編碼解碼

一，netty+protocol buffers簡要說明

Netty是業界最流行的NIO框架之一
優勢：
1)API使用簡單，開發門檻低；
2)功能強大，預置了多種編解碼功能，支持多種主流協議；
3)定製能力強，能夠經過ChannelHandler對通訊框架進行靈活的擴展；
4)性能高，經過與其它業界主流的NIO框架對比，Netty的綜合性能最優；
5)成熟、穩定，Netty修復了已經發現的全部JDK NIO BUG，業務開發人員不須要再爲NIO的BUG而煩惱；
6)社區活躍，版本迭代週期短，發現的BUG能夠被及時修復，同時，更多的新功能會被加入；
7)經歷了大規模的商業應用考驗，質量已經獲得驗證。在互聯網、大數據、網絡遊戲、企業應用、電信軟件等衆多行業獲得成功商用，證實了它能夠徹底知足不一樣行業的商業應用。

 

Protocol Buffers是Google公司開發的一種以有效並可擴展的格式編碼結構化數據的方式。
可用於數據存儲、通訊協議等方面,它不依賴於語言和平臺而且可擴展性極強。
現階段官方支持C++、JAVA、Python等三種編程語言，但能夠找到大量的幾乎涵蓋全部語言的第三方拓展包。

優勢
1)二進制消息，性能好/效率高（空間和時間效率都很不錯）
2)proto文件生成目標代碼，簡單易用
3)序列化反序列化直接對應程序中的數據類，不須要解析後在進行映射(XML,JSON都是這種方式)
4)支持向前兼容（新加字段採用默認值）和向後兼容（忽略新加字段），簡化升級
5)支持多種語言

 二，認識varint

proto 消息格式如：Length + Protobuf Data (消息頭+消息數據)

消息頭描述消息數據體的長度。爲了更減小傳輸量，消息頭採用的是varint 格式。

什麼是varint?

Varint 是一種緊湊的表示數字的方法。它用一個或多個字節來表示一個數字，值越小的數字使用越少的字節數。這能減小用來表示數字的字節數。
Varint 中的每一個 byte 的最高位 bit 有特殊的含義，若是該位爲 1，表示後續的 byte 也是該數字的一部分，若是該位爲 0，則結束。其餘的 7 個 bit 都用來表示數字。所以小於 128 的數字均可以用一個 byte 表示。大於 128 的數字，會用兩個字節。

例如整數1的表示，僅需一個字節：

0000 0001

例如300的表示，須要兩個字節：

1010 1100 0000 0010

採 用 Varint，對於很小的 int32 類型的數字，則能夠用 1 個 byte 來表示。固然凡事都有好的也有很差的一面，採用 Varint 表示法，大的數字則須要 5 個 byte 來表示。從統計的角度來講，通常不會全部的消息中的數字都是大數，所以大多數狀況下，採用 Varint 後，能夠用更少的字節數來表示數字信息。

下圖演示了 Google Protocol Buffer 如何解析兩個 bytes。注意到最終計算前將兩個 byte 的位置相互交換過一次，這是由於 Google Protocol Buffer 字節序採用 little-endian 的方式。

三，通訊的編碼解碼

netty 默認提供了對protocol buffers 的支持，因此整合起來很簡單。整合的關係，關鍵在於對編碼解碼的理解。

2.11 服務器端起用一個服務基本的代碼以下：

EventLoopGroup boosGroup = new NioEventLoopGroup();
		EventLoopGroup workGroup = new NioEventLoopGroup();

		try {
			ServerBootstrap bootstrap = new ServerBootstrap();
			bootstrap.group(boosGroup, workGroup).channel(NioServerSocketChannel.class)
					.option(ChannelOption.SO_BACKLOG, 100).childHandler(new ChannelInitializer<Channel>() {

						@Override
						protected void initChannel(Channel ch) throws Exception {

							ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());// 解碼(處理半包)
							ch.pipeline().addLast(new ProtobufDecoder(MsgProto.Packet.getDefaultInstance()));
							ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());//加長度
							ch.pipeline().addLast(new ProtobufEncoder());// 編碼

							ch.pipeline().addLast(new ServerChannelHandlerAdapter());// 業務處理handler
						}

					});
			// 綁定端口
			ChannelFuture future = bootstrap.bind(2015).sync();
			// 等待關閉
			future.channel().closeFuture().sync();

		} catch (Exception e) {
			LOG.error("{}", e);
		} finally {
			boosGroup.shutdownGracefully();
			workGroup.shutdownGracefully();
		}

　　

2.21 重點查看編碼解碼源碼類，ProtobufVarint32FrameDecoder，ProtobufVarint32LengthFieldPrepender，ProtobufEncoder

編碼類：

ProtobufEncoder：這裏只是把proto 對象直接寫入out

@Sharable
public class ProtobufEncoder extends MessageToMessageEncoder<MessageLiteOrBuilder> {
    @Override
    protected void encode(
            ChannelHandlerContext ctx, MessageLiteOrBuilder msg, List<Object> out) throws Exception {
        if (msg instanceof MessageLite) {
            out.add(wrappedBuffer(((MessageLite) msg).toByteArray()));
            return;
        }
        if (msg instanceof MessageLite.Builder) {
            out.add(wrappedBuffer(((MessageLite.Builder) msg).build().toByteArray()));
        }
    }
}

關鍵是ProtobufVarint32LengthFieldPrepender類：

* BEFORE DECODE (300 bytes) AFTER DECODE (302 bytes)
* +---------------+ +--------+---------------+
* | Protobuf Data |-------------->| Length | Protobuf Data |
* | (300 bytes) | | 0xAC02 | (300 bytes) |
* +---------------+ +--------+---------------+

//數據格式爲=數據長度（頭部）+ 真正的數據
@Sharable
public class ProtobufVarint32LengthFieldPrepender extends MessageToByteEncoder<ByteBuf> {

    @Override
    protected void encode(
            ChannelHandlerContext ctx, ByteBuf msg, ByteBuf out) throws Exception {
        int bodyLen = msg.readableBytes();
        int headerLen = CodedOutputStream.computeRawVarint32Size(bodyLen);
        out.ensureWritable(headerLen + bodyLen);//確保可寫長度（頭部長度+數據長度）
        //數據長度寫入頭部
        CodedOutputStream headerOut =
                CodedOutputStream.newInstance(new ByteBufOutputStream(out), headerLen);
        headerOut.writeRawVarint32(bodyLen);
        headerOut.flush();
        //寫數據
        out.writeBytes(msg, msg.readerIndex(), bodyLen);
    }
}

　　

 

對應的解碼類 ProtobufVarint32FrameDecoder：

作的事情如類註解：

* BEFORE DECODE (302 bytes) AFTER DECODE (300 bytes)
* +--------+---------------+ +---------------+
* | Length | Protobuf Data |----->| Protobuf Data |
* | 0xAC02 | (300 bytes) | | (300 bytes) |
* +--------+---------------+ +---------------+

//半包粘包處理
public class ProtobufVarint32FrameDecoder extends ByteToMessageDecoder {

    // TODO maxFrameLength + safe skip + fail-fast option
    //      (just like LengthFieldBasedFrameDecoder)

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        in.markReaderIndex();//標記讀取的位置
        final byte[] buf = new byte[5];//varint32 最大5個字節
        for (int i = 0; i < buf.length; i ++) {
            if (!in.isReadable()) {
                in.resetReaderIndex();
                return;
            }

            buf[i] = in.readByte();
            if (buf[i] >= 0) {
                int length = CodedInputStream.newInstance(buf, 0, i + 1).readRawVarint32();//varint表示的格式 轉 實際長度int
                if (length < 0) {
                    throw new CorruptedFrameException("negative length: " + length);
                }

                if (in.readableBytes() < length) {//長度不夠，回滾標記
                    in.resetReaderIndex();
                    return;
                } else {
                    out.add(in.readBytes(length));//正確讀取返回
                    return;
                }
            }
        }

        // Couldn't find the byte whose MSB is off.
        throw new CorruptedFrameException("length wider than 32-bit");
    }
}

　　

 

 

四，總結

 

netty 默認已經幫咱們實現了protocol buffers 的編碼解碼，因此使用起來很方便。

但若是有特殊須要，如加密等，本身定義編碼解碼則須要理解編碼解碼的規則和參考它的默認實現。