咨询gRPC中bytes类型的用途、适用场景及示例

核心用途 #

gRPC的Protobuf定义里，bytes类型是专门用来传输原始二进制数据的通用容器。和string、int等结构化类型不同，它不会对数据做任何编码或格式转换，直接按字节流传递，完美适配非结构化、预序列化或加密后的二进制内容，填补了结构化类型无法高效处理二进制数据的空白。

适用场景与使用时机 #

你可以在以下场景优先选择bytes类型：

• 传输非结构化二进制文件：比如图片、音频、视频片段，或者压缩包（ZIP/GZIP）、PDF等。这类数据本身就是二进制格式，用bytes传输无需额外编码（比如转Base64），既节省带宽也避免编码错误。
• 自定义序列化的数据：如果你不想用Protobuf自带的序列化机制，而是采用MsgPack、JSON序列化后的字节流，甚至是自定义的二进制协议，bytes可以直接承载这些预序列化的内容，灵活对接不同的序列化方案。
• 加密/敏感数据：加密后的密文是二进制形式，直接用bytes传输无需转码，减少数据处理环节的开销，同时降低编码过程中泄露敏感信息的风险。
• 对接异构/旧系统：如果需要和不支持Protobuf的旧系统或异构服务交互，对方输出的是二进制数据，用bytes可以直接传递，无需复杂的格式转换。

代码示例 #

1. Protobuf定义文件 #

首先在.proto文件中定义使用bytes的服务和消息：

syntax = "proto3";

service BinaryDataService {
  // 上传文件片段
  rpc UploadFileChunk (FileChunk) returns (ChunkResponse);
  // 传输自定义序列化数据
  rpc ExchangeCustomData (CustomBinaryData) returns (CustomBinaryData);
}

message FileChunk {
  string file_unique_id = 1;
  bytes chunk_content = 2; // 存储文件二进制片段
  int32 chunk_number = 3;
}

message ChunkResponse {
  bool is_success = 1;
  string msg = 2;
}

message CustomBinaryData {
  bytes serialized_payload = 1; // 存储自定义序列化后的字节数据
}

2. Python服务端实现 #

import grpc
from concurrent import futures
import binary_data_pb2
import binary_data_pb2_grpc
import msgpack

class BinaryDataServicer(binary_data_pb2_grpc.BinaryDataServiceServicer):
    def UploadFileChunk(self, request, context):
        # 将收到的二进制片段写入临时文件
        with open(f"{request.file_unique_id}.part", "ab") as f:
            f.write(request.chunk_content)
        return binary_data_pb2.ChunkResponse(
            is_success=True,
            msg=f"Chunk {request.chunk_number} saved successfully"
        )

    def ExchangeCustomData(self, request, context):
        # 用MsgPack反序列化收到的字节数据
        raw_data = msgpack.unpackb(request.serialized_payload)
        # 处理数据：添加标记
        raw_data["processed"] = True
        # 重新序列化后返回
        processed_bytes = msgpack.packb(raw_data)
        return binary_data_pb2.CustomBinaryData(serialized_payload=processed_bytes)

def start_server():
    server = grpc.server(futures.ThreadPoolExecutor(max_workers=10))
    binary_data_pb2_grpc.add_BinaryDataServicer_to_server(BinaryDataServicer(), server)
    server.add_insecure_port("[::]:50051")
    server.start()
    server.wait_for_termination()

if __name__ == "__main__":
    start_server()

3. Python客户端实现 #

import grpc
import binary_data_pb2
import binary_data_pb2_grpc
import msgpack

def upload_image_chunk():
    # 读取本地图片的1MB片段
    with open("demo.png", "rb") as f:
        chunk = f.read(1024 * 1024)
    
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = binary_data_pb2_grpc.BinaryDataServiceStub(channel)
        response = stub.UploadFileChunk(binary_data_pb2.FileChunk(
            file_unique_id="demo_png_123",
            chunk_content=chunk,
            chunk_number=0
        ))
        print(response.msg)

def send_custom_serialized_data():
    # 构造原始数据，用MsgPack序列化
    user_data = {"username": "john_doe", "email": "john@example.com"}
    serialized_data = msgpack.packb(user_data)
    
    with grpc.insecure_channel("localhost:50051") as channel:
        stub = binary_data_pb2_grpc.BinaryDataServiceStub(channel)
        response = stub.ExchangeCustomData(binary_data_pb2.CustomBinaryData(
            serialized_payload=serialized_data
        ))
        # 反序列化服务端返回的处理后数据
        processed_data = msgpack.unpackb(response.serialized_payload)
        print("Processed Data:", processed_data)

if __name__ == "__main__":
    upload_image_chunk()
    send_custom_serialized_data()

注意事项 #

• bytes类型在传输时不会做任何编码处理，所以如果是自定义二进制协议，需要自己处理跨平台的字节序问题；
• gRPC默认有消息大小限制（一般是4MB），如果传输大文件，务必像示例一样分块传输，避免超出限制；
• 对于文本类数据，优先用string类型，bytes更适合真正的二进制内容，不要混用。

内容的提问来源于stack exchange，提问作者PRAJAKTA DHIVAR