Dubbo3之Triple协议的消息序列化

前言

在 RPC 调用中，消息的打包和解包是实现数据的传输和交互的关键步骤之一。
当客户端发起一次 RPC 调用时，客户端需要将调用的相关参数打包成一个消息，然后将消息发送给服务端。
服务端接收到消息后，需要对消息进行解包，提取出参数进行处理，最后将处理结果打包成响应消息发送回客户端。

消息打包是将调用参数和其他相关信息组装成一个字节序列的过程，即序列化。
消息解包是将接收到的字节流还原为原始的调用参数和相关信息的过程，即反序列化。

Pack & UnPack

在 Dubbo3 中，需要打包和解包的消息主要是 Request 和 Response。
又因为 Triple 协议除了要支持传输 Protobuf 消息，还要能传输普通的pojo类。所以再按照是否需要包装来区分，Dubbo3 一共提供了如下类：

Pack 是消息打包的接口：

interface Pack {
    byte[] pack(Object obj) throws IOException;
}

UnPack 是消息解包的接口：

interface UnPack {
    Object unpack(byte[] data) throws IOException, ClassNotFoundException;
}

实现类中，Pb开头的类是专门针对Protobuf消息的，Wrap开头的类针对的是普通的pojo类。

消息Wrap

Dubbo 怎么判断消息打包是否需要Wrap？？？
Triple 协议要传输的消息有两种：由Protobuf插件编译好的消息、普通pojo类。
前者自带打包和解包的能力，所以不需要 Dubbo 额外处理，即无需Wrap。Protobuf 消息也很好判断，就是判断类是否实现了com.google.protobuf.Message接口。

Dubbo 又是如何Wrap消息的呢？？？
普通的 pojo 类是没有序列化能力的，该如何按照Protobuf的方式序列化传输呢？
Dubbo 会把普通的 pojo 类统一包装成 TripleRequestWrapper 和 TripleResponseWrapper。
对应的proto文件路径：dubbo-rpc/dubbo-rpc-triple/src/main/proto/triple_wrapper.proto。

syntax = "proto3";

package org.apache.dubbo.triple;

message TripleRequestWrapper {
    // hessian4
    // json
    string serializeType = 1;
    repeated bytes args = 2;
    repeated string argTypes = 3;
}

message TripleResponseWrapper {
    string serializeType = 1;
    bytes data = 2;
    string type = 3;
}

message TripleExceptionWrapper {
    string language = 1;
    string serialization = 2;
    string className = 3;
    bytes data = 4;
}

字段 serializeType 仍然可以指定序列化方式，意味着你的参数部分仍然可以使用 hessian、json 等序列化方式，但是完整的请求-响应消息必须使用Protobuf序列化。

序列化

如果你了解 Protobuf 序列化的规则，就很容易理解Wrap消息的序列化代码。
以 Request 为例，打包方法是WrapRequestPack#pack()：

public byte[] pack(Object obj) throws IOException {
    Object[] arguments;
    if (singleArgument) {
        arguments = new Object[]{obj};
    } else {
        arguments = (Object[]) obj;
    }
    // 包装成 TripleRequestWrapper 消息
    final TripleCustomerProtocolWapper.TripleRequestWrapper.Builder builder = TripleCustomerProtocolWapper.TripleRequestWrapper.Builder.newBuilder();
    // 序列化类型
    builder.setSerializeType(serialize);
    for (String type : argumentsType) {
        // 形参类型
        builder.addArgTypes(type);
    }
    // 按顺序序列化 -> List
    ByteArrayOutputStream bos = new ByteArrayOutputStream();
    for (int i = 0; i < arguments.length; i++) {
        Object argument = arguments[i];
        multipleSerialization.serialize(url, serialize, actualRequestTypes[i], argument, bos);
        builder.addArgs(bos.toByteArray());
        bos.reset();
    }
    // protobuf 序列化
    return builder.build().toByteArray();
}

上面一步只是构建好了 TripleRequestWrapper 对象，对象本身是不能传输的，还要把它按照 Protobuf 的方式系列化成字节序列。
方法是：TripleRequestWrapper#toByteArray()：

public byte[] toByteArray() {
    int totalSize = 0;
    // 生成tag 字段顺序是1 wireType=2
    int serializeTypeTag = makeTag(1, 2);
    // tag varint编码 可变长度整型
    byte[] serializeTypeTagBytes = varIntEncode(serializeTypeTag);
    byte[] serializeTypeBytes = serializeType.getBytes(StandardCharsets.UTF_8);
    // wireType=2 变长类型 需要记录length 也是varint
    byte[] serializeTypeLengthVarIntEncodeBytes = varIntEncode(serializeTypeBytes.length);
    totalSize += serializeTypeTagBytes.length
        + serializeTypeLengthVarIntEncodeBytes.length
        + serializeTypeBytes.length;
    int argTypeTag = makeTag(3, 2);
    if (CollectionUtils.isNotEmpty(argTypes)) {
        totalSize += varIntComputeLength(argTypeTag) * argTypes.size();
        for (String argType : argTypes) {
            byte[] argTypeBytes = argType.getBytes(StandardCharsets.UTF_8);
            totalSize += argTypeBytes.length + varIntComputeLength(argTypeBytes.length);
        }
    }
    int argTag = makeTag(2, 2);
    if (CollectionUtils.isNotEmpty(args)) {
        totalSize += varIntComputeLength(argTag) * args.size();
        for (byte[] arg : args) {
            totalSize += arg.length + varIntComputeLength(arg.length);
        }
    }
    ByteBuffer byteBuffer = ByteBuffer.allocate(totalSize);
    byteBuffer
        .put(serializeTypeTagBytes)
        .put(serializeTypeLengthVarIntEncodeBytes)
        .put(serializeTypeBytes);
    if (CollectionUtils.isNotEmpty(args)) {
        byte[] argTagBytes = varIntEncode(argTag);
        for (byte[] arg : args) {
            byteBuffer
                .put(argTagBytes)
                .put(varIntEncode(arg.length))
                .put(arg);
        }
    }
    if (CollectionUtils.isNotEmpty(argTypes)) {
        byte[] argTypeTagBytes = varIntEncode(argTypeTag);
        for (String argType : argTypes) {
            byte[] argTypeBytes = argType.getBytes(StandardCharsets.UTF_8);
            byteBuffer
                .put(argTypeTagBytes)
                .put(varIntEncode(argTypeBytes.length))
                .put(argTypeBytes);
        }
    }
    return byteBuffer.array();
}

反序列化的代码是WrapRequestUnpack#unpack()，就是pack()的逆向流程：

public Object unpack(byte[] data) throws IOException, ClassNotFoundException {
    // Protobuf方式 反序列化为 TripleRequestWrapper
    TripleCustomerProtocolWapper.TripleRequestWrapper wrapper = TripleCustomerProtocolWapper.TripleRequestWrapper.parseFrom(
        data);
    // 序列化类型
    String wrapperSerializeType = convertHessianFromWrapper(wrapper.getSerializeType());
    CodecSupport.checkSerialization(serializeName, wrapperSerializeType);
    // 实参反序列化
    Object[] ret = new Object[wrapper.getArgs().size()];
    ((WrapResponsePack) responsePack).serialize = wrapper.getSerializeType();
    for (int i = 0; i < wrapper.getArgs().size(); i++) {
        ByteArrayInputStream bais = new ByteArrayInputStream(
            wrapper.getArgs().get(i));
        ret[i] = serialization.deserialize(url, wrapper.getSerializeType(),
            actualRequestTypes[i],
            bais);
    }
    return ret;
}

核心是TripleRequestWrapper#parseFrom()，它会按照 Protobuf 的规则将字节序列重新解包成 TripleRequestWrapper 对象。

public static TripleRequestWrapper parseFrom(byte[] data) {
    TripleRequestWrapper tripleRequestWrapper = new TripleRequestWrapper();
    ByteBuffer byteBuffer = ByteBuffer.wrap(data);
    tripleRequestWrapper.args = new ArrayList();
    tripleRequestWrapper.argTypes = new ArrayList();
    // 循环读
    while (byteBuffer.position() < byteBuffer.limit()) {
        // 读第一个tag
        int tag = readRawVarint32(byteBuffer);
        // 字段序号
        int fieldNum = extractFieldNumFromTag(tag);
        // wireType 必须是2 因为都是变长类型
        int wireType = extractWireTypeFromTag(tag);
        if (wireType != 2) {
            throw new RuntimeException(String.format("unexpect wireType, expect %d realType %d", 2, wireType));
        }
        if (fieldNum == 1) {// serializeType
            int serializeTypeLength = readRawVarint32(byteBuffer);
            byte[] serializeTypeBytes = new byte[serializeTypeLength];
            byteBuffer.get(serializeTypeBytes, 0, serializeTypeLength);
            tripleRequestWrapper.serializeType = new String(serializeTypeBytes);
        } else if (fieldNum == 2) {// args
            int argLength = readRawVarint32(byteBuffer);
            byte[] argBytes = new byte[argLength];
            byteBuffer.get(argBytes, 0, argLength);
            tripleRequestWrapper.args.add(argBytes);
        } else if (fieldNum == 3) {// argTypes
            int argTypeLength = readRawVarint32(byteBuffer);
            byte[] argTypeBytes = new byte[argTypeLength];
            byteBuffer.get(argTypeBytes, 0, argTypeLength);
            tripleRequestWrapper.argTypes.add(new String(argTypeBytes));
        } else {
            throw new RuntimeException("fieldNum should in (1,2,3)");
        }
    }
    return tripleRequestWrapper;
}

尾巴

消息打包是将数据结构转换为字节流的过程，而消息解包则是将字节流转换回原始数据结构的过程。
这两个过程在RPC调用中起着至关重要的作用，因为它们使得不同系统之间可以通过网络传输数据，并使得远程过程调用成为可能。
Dubbo3 的 Triple 协议，除了要支持传输 Protobuf 消息，还要能传输普通的 pojo 类，因为对于多语言诉求不强的公司来说，强制使用 IDL 来定义服务成本太高了。针对 pojo 类，Dubbo 会把它们统一包装成Wrapper，再按照 Protobuf 的方式序列化传输，对端再按照相同的方式反序列化即可。