您现在的位置：程式碼之家 >> 編程語言 > >> Java 程式設計 >> Java 綜合教學

Protobuf3 與 Netty4 實現 Socket 多類型數據傳輸：自定義編解碼器方案詳解

編輯：Java 綜合教學

Protobuf3 與 Netty4 實現 Socket 多類型數據傳輸：自定義編解碼器方案詳解

Protobuf 序列化後的位元組流不具備自描述能力，接收端必須預先知道數據類型才能正確反序列化。這嚴重限制了單一連接通道傳輸多種數據類型的能力。本文詳細介紹一種高效的自定義協議頭方案，在保持高效能的前提下實現多類型 Protobuf 數據的混合傳輸。

方案一：官方 SelfDescribingMessage（不推薦用於高效能場景）

Protobuf 官方 Wiki 提供了一種替代方案，定義一個包含元數據的封裝訊息類型：

message SelfDescribingMessage {
  required FileDescriptorSet proto_files = 1;
  required string type_name = 2;
  required bytes message_data = 3;
}

實際數據放在 message_data，類型資訊放在 proto_files 和 type_name。雖然可行，但需要兩次序列化/反序列化，位元組陣列建立兩次，對延遲敏感系統不夠理想。

方案二：自定義協議頭 + Netty 編解碼器（推薦）

核心思路：在 Protobuf 序列化數據前加上自定義協議頭，包含數據長度與類型標識。解碼時根據協議頭資訊進行反序列化。

場景定義

假設需傳輸兩類行情數據：股票行情 (StockTick) 與期權行情 (OptionTick)。

// StockTick.proto
syntax = "proto3";
package test.model.protobuf;
option java_package = "test.model.protobuf";
message StockTick {
  string stockId = 1;
  int price = 2;
}

// OptionTick.proto
syntax = "proto3";
package test.model.protobuf;
option java_package = "test.model.protobuf";
message OptionTick {
  string optionId = 1;
  string securityId = 2;
  int price = 3;
}

Netty 官方編解碼器限制

Netty 內建的 ProtobufDecoder 必須指定單一 Protobuf 類型，無法直接支援多類型混合傳輸：

pipeline.addLast(new ProtobufVarint32FrameDecoder());
pipeline.addLast(new ProtobufDecoder(StockTickOuterClass.StockTick.getDefaultInstance())); // 固定單一類型
pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
pipeline.addLast(new ProtobufEncoder());
pipeline.addLast(new CustomProtoServerHandler());

自定義編碼器：CustomProtobufEncoder

@Sharable
public class CustomProtobufEncoder extends MessageToByteEncoder<MessageLite> {
    private final HangqingEncoder hangqingEncoder;
    public CustomProtobufEncoder(HangqingEncoder hangqingEncoder) {
        this.hangqingEncoder = hangqingEncoder;
    }
    @Override
    protected void encode(ChannelHandlerContext ctx, MessageLite msg, ByteBuf out) throws Exception {
        byte[] body = msg.toByteArray();
        byte[] header = encodeHeader(msg, (short) body.length);
        out.writeBytes(header);
        out.writeBytes(body);
    }
    private byte[] encodeHeader(MessageLite msg, short bodyLength) {
        byte messageType = 0x0F; // 預設未知類型
        if (msg instanceof StockTickOuterClass.StockTick) {
            messageType = 0x00;
        } else if (msg instanceof OptionTickOuterClass.OptionTick) {
            messageType = 0x01;
        }
        // 協議頭格式：body長度低位 | body長度高位 | 保留字節 | 類型標識
        byte[] header = new byte[4];
        header[0] = (byte) (bodyLength & 0xFF);
        header[1] = (byte) ((bodyLength >> 8) & 0xFF);
        header[2] = 0; // 保留字節
        header[3] = messageType;
        return header;
    }
}

自定義解碼器：CustomProtobufDecoder

public class CustomProtobufDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        while (in.readableBytes() > 4) {
            in.markReaderIndex();
            // 讀取 body 長度（小端序）
            byte low = in.readByte();
            byte high = in.readByte();
            short length = (short) (((high & 0xFF) << 8) | (low & 0xFF));
            // 跳過保留字節，讀取類型標識
            in.readByte(); // reserved
            byte dataType = in.readByte();
            // 半包檢查
            if (in.readableBytes() < length) {
                in.resetReaderIndex();
                return;
            }
            // 讀取 body
            ByteBuf bodyByteBuf = in.readBytes(length);
            byte[] array;
            int offset;
            int readableLen = bodyByteBuf.readableBytes();
            if (bodyByteBuf.hasArray()) {
                array = bodyByteBuf.array();
                offset = bodyByteBuf.arrayOffset() + bodyByteBuf.readerIndex();
            } else {
                array = new byte[readableLen];
                bodyByteBuf.getBytes(bodyByteBuf.readerIndex(), array, 0, readableLen);
                offset = 0;
            }
            // 根據類型標識反序列化
            MessageLite result = decodeBody(dataType, array, offset, readableLen);
            if (result != null) {
                out.add(result);
            }
        }
    }
    private MessageLite decodeBody(byte dataType, byte[] array, int offset, int length) throws Exception {
        switch (dataType) {
            case 0x00:
                return StockTickOuterClass.StockTick.getDefaultInstance().getParserForType().parseFrom(array, offset, length);
            case 0x01:
                return OptionTickOuterClass.OptionTick.getDefaultInstance().getParserForType().parseFrom(array, offset, length);
            default:
                return null; // 或拋出異常
        }
    }
}

在 Netty Server 中應用

@Override
protected void initChannel(SocketChannel ch) throws Exception {
    ChannelPipeline pipeline = ch.pipeline();
    pipeline.addLast("decoder", new CustomProtobufDecoder());
    pipeline.addLast("encoder", new CustomProtobufEncoder());
    pipeline.addLast(new CustomProtoServerHandler());
}