如何用Python通过gRPC以4MB块流式传输数据及相关最佳实践

一、4MB分块发送的实现 #

根据TCP限制，将数据按4MB（4194304字节）切分后逐个发送，现有proto定义下的实现代码如下：

基础实现（一次性序列化后分块） #

适合数据量不是极端大的场景：

def GetGeoCoordinates(self, request, context):
    # 假设coordinates["data"]是你的原始大数据集合
    ...
    ...

    # 1. 将数据转为JSON字节流（兼容现有逻辑）
    json_bytes = json.dumps(coordinates["data"]).encode('utf-8')
    # 2. 定义4MB块大小
    CHUNK_SIZE = 4 * 1024 * 1024  # 4194304字节
    # 3. 循环切分并流式发送
    for start in range(0, len(json_bytes), CHUNK_SIZE):
        end = start + CHUNK_SIZE
        chunk_data = json_bytes[start:end]
        yield geo_pb2.Chunk(data_part=chunk_data)

内存优化实现（增量序列化分块） #

如果数据量极大，一次性生成完整JSON字节流会占用过多内存，可以用ijson库增量序列化，避免内存溢出：

import ijson

def GetGeoCoordinates(self, request, context):
    CHUNK_SIZE = 4 * 1024 * 1024
    buffer = bytearray()

    # 增量遍历数据并序列化，避免一次性加载全量数据
    for prefix, _, value in ijson.items(coordinates["data"], ''):
        # 将当前数据片段转为JSON字节并加入缓冲区
        buffer.extend(json.dumps({prefix: value}).encode('utf-8'))
        # 缓冲区达到块大小就发送
        while len(buffer) >= CHUNK_SIZE:
            yield geo_pb2.Chunk(data_part=buffer[:CHUNK_SIZE])
            buffer = buffer[CHUNK_SIZE:]
    # 发送剩余的不足一块的数据
    if buffer:
        yield geo_pb2.Chunk(data_part=buffer)

二、先json.dumps再流式传输是否为最佳实践？ #

不是最优选择，核心问题如下：

• 内存开销高：一次性将全量数据转为JSON字节流，会在内存中占用与原始数据相当甚至更大的空间，数据量极大时易触发内存告警。
• 序列化效率低：JSON的序列化/反序列化性能远低于Protobuf的结构化序列化，复杂数据结构下差距更明显。
• 无类型安全：JSON是弱类型格式，客户端解析时容易出现类型不匹配的错误，而Protobuf的强类型定义可避免此类问题。

更优的实践方案 #

1. 若可修改Proto定义（推荐）

直接用Protobuf定义地理坐标的结构化消息，分块发送结构化数据，无需转JSON：

service GeoService {
   rpc GetGeoCoordinates(GetRequest) returns (stream Chunk){}
}

// 新增地理坐标结构化消息
message GeoCoordinate {
    double latitude = 1;
    double longitude = 2;
    // 其他业务字段...
}

message Chunk {
    // 用结构化列表替代bytes，效率更高且类型安全
    repeated GeoCoordinate coordinates = 1;
}

服务端实现：

def GetGeoCoordinates(self, request, context):
    # 每块包含1000个坐标（可根据字节大小调整）
    CHUNK_ITEM_COUNT = 1000
    coords_list = coordinates["data"]  # 假设是GeoCoordinate对象列表
    for start in range(0, len(coords_list), CHUNK_ITEM_COUNT):
        end = start + CHUNK_ITEM_COUNT
        yield geo_pb2.Chunk(coordinates=coords_list[start:end])

2. 若不可修改现有Proto定义

将结构化数据序列化为Protobuf字节流，再分块发送，比JSON更高效：

# 先定义一个包含坐标列表的Protobuf消息（仅服务端和客户端内部使用）
# message GeoCoordinateList { repeated GeoCoordinate coords = 1; }

def GetGeoCoordinates(self, request, context):
    # 将原始数据转为Protobuf结构化消息
    coords_pb = geo_pb2.GeoCoordinateList(coords=coordinates["data"])
    proto_bytes = coords_pb.SerializeToString()
    CHUNK_SIZE = 4 * 1024 * 1024
    # 分块发送Protobuf字节流
    for start in range(0, len(proto_bytes), CHUNK_SIZE):
        end = start + CHUNK_SIZE
        yield geo_pb2.Chunk(data_part=proto_bytes[start:end])

客户端收到所有块后合并字节流，再反序列化为GeoCoordinateList即可。

内容的提问来源于stack exchange，提问作者ABDULLOKH MUKHAMMADJONOV