上传文件至GCP时如何确定记录大小，解决超10MB限制的报错问题

你遇到的报错实际来自GCP Pub/Sub服务，通常是上传Blob后触发Pub/Sub通知、或者上传前需要发布对应业务消息到Pub/Sub时触发的大小限制，报错翻译如下：

google.cloud.pubsub_v1.publisher.exceptions.MessageTooLargeError：待发布的消息生成的发布请求过大，超出了后端允许的最大大小（10000000字节，即10MB）

优化思路 #

• 在调用上传、发布逻辑前，预先计算待处理记录序列化后的大小，提前拦截超限制内容，避免触发服务端报错
• 显式捕获指定异常，对超出限制的记录做落日志、分块、单独存储等自定义处理
• 注意GCP Pub/Sub单条消息的官方最大限制确实为10MB，如果你是上传后触发Pub/Sub通知，要额外确认上传的元数据、消息负载总大小是否超出限制

Python脚本优化示例 #

from google.cloud import storage, pubsub_v1
from google.cloud.pubsub_v1.publisher.exceptions import MessageTooLargeError
import json

# 常量定义：GCP Pub/Sub单条消息最大限制 10MB
MAX_PUBSUB_MESSAGE_SIZE = 10 * 1024 * 1024  # 对应10000000字节

# 初始化客户端，替换为你自己的项目、桶、Topic参数
storage_client = storage.Client()
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path("你的GCP项目ID", "你的Pub/Sub Topic名称")
bucket = storage_client.bucket("你的GCS存储桶名称")

def process_single_record(record):
    # 先序列化待发送的消息内容，计算实际大小
    message_payload = json.dumps(record).encode("utf-8")
    payload_size = len(message_payload)

    # 预校验：提前识别超出大小的记录
    if payload_size > MAX_PUBSUB_MESSAGE_SIZE:
        # 这里可以自定义处理逻辑：比如记录日志、拆分记录、存储到单独的超大记录目录等
        print(f"记录超出大小限制，预估大小：{payload_size}字节，最大允许：{MAX_PUBSUB_MESSAGE_SIZE}字节，记录标识：{record.get('id', '未知标识')}")
        # 可选择返回跳过，或者走分块处理逻辑
        return False
    
    # 原有上传+发布逻辑
    try:
        # 上传Blob到GCS
        blob = bucket.blob(f"records/{record['id']}.json")
        blob.upload_from_string(message_payload, content_type="application/json")
        
        # 发布Pub/Sub消息
        future = publisher.publish(topic_path, message_payload, blob_path=blob.name)
        future.result()
        return True
    except MessageTooLargeError as e:
        # 兜底捕获异常，覆盖预校验遗漏的特殊场景
        print(f"捕获到消息过大异常，记录标识：{record.get('id', '未知标识')}，错误信息：{str(e)}")
        # 自定义异常处理，比如把超大记录单独转存到指定目录
        blob = bucket.blob(f"oversized_records/{record['id']}.json")
        blob.upload_from_string(message_payload, content_type="application/json")
        return False
    except Exception as e:
        print(f"处理记录异常，记录标识：{record.get('id', '未知标识')}，错误信息：{str(e)}")
        return False

额外可选优化点 #

• 如果业务允许，可对超出大小的记录做分块拆分，拆成多条小于10MB的消息分别发送，上传的Blob也对应拆分为分片对象
• 超大记录可单独存储到GCS后，Pub/Sub消息仅传递超大记录的GCS路径，避免传递完整大负载
• 可将所有超限制的记录信息统一落表存储，方便后续批量重处理

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