如何提升Cloud Run中GCS Blob上传吞吐量以解决PubSub推送订阅超时问题

先揪出你代码里最核心的问题：线程池的使用逻辑完全错误，这直接导致你根本没用到多线程并行上传，是吞吐量上不去的关键原因。除此之外还有几个可以优化的点，逐个给你拆解：

1. 修复线程池的使用方式 #

你在extract_frames里用executor.map(upload, (blob, buf))是典型的用法错误——map会把传入的可迭代对象的每个元素单独传给upload函数，也就是说第一次会把blob传给upload的第一个参数，第二次把buf传给第一个参数，完全不符合upload需要两个参数的要求，而且每次循环只提交了一个无效任务，本质还是串行上传。

正确的做法是用executor.submit()提交每个上传任务，同时设置合适的线程池大小（GCS上传是IO密集型任务，线程数可以远高于CPU核数，比如20-50）：

修改extract_frames函数：

def extract_frames(
    signed_url : str,
    basename : str,
    pathname : str,
    dst_bucket_name : str = "extracted-frames"
) -> int:
    dst_client = storage.Client()
    dst_bucket = dst_client.get_bucket(dst_bucket_name)
    count = 0
    vid = cv2.VideoCapture(signed_url)
    # 针对IO密集型任务设置合理的线程数
    with ThreadPoolExecutor(max_workers=30) as executor:
        futures = []  # 保存任务对象，可选：等待所有上传完成再返回
        ret,frame = vid.read()
        while ret:
            enc_ret, buf = cv2.imencode(".jpg", frame)
            if not enc_ret:
                msg = f'Bad Encoding [Frame: {count:06}]'
                print(msg)
            else:
                blob_name = f"{pathname}/{basename}-{count:06}.jpg"
                blob = dst_bucket.blob(blob_name)
                # 用submit传入正确的两个参数，提交上传任务
                futures.append(executor.submit(upload, blob, buf))
            count += 1
            ret,frame = vid.read()
        # 可选：等待所有上传任务完成，避免函数提前结束导致任务中断
        for future in futures:
            future.result()
    vid.release()
    return count

2. 复用GCS客户端，减少初始化开销 #

你现在每次处理任务都会创建新的src_client和dst_client，但GCS客户端是线程安全的，可以全局复用，不需要重复初始化，能省不少资源开销：

在video.py顶部全局初始化客户端：

import os
from datetime import timedelta
from concurrent.futures import ThreadPoolExecutor
import cv2
from google.cloud import storage
from google.oauth2 import service_account

# 全局复用GCS客户端，避免重复初始化
src_client = storage.Client()
dst_client = storage.Client()
dst_bucket = dst_client.get_bucket("extracted-frames")

def upload(blob : storage.blob.Blob, buf : "numpy.ndarray"):
    blob.upload_from_string(buf.tobytes(), content_type="image/jpeg")

def process(data : dict):
    src_bucket = src_client.get_bucket(data["bucket"])
    src_blob = src_bucket.get_blob(data["name"])
    pathname = os.path.dirname(data["name"])
    basename, ext = os.path.splitext(os.path.basename(data["name"]))
    # 移除本地密钥文件依赖，用Cloud Run默认凭据
    url = src_blob.generate_signed_url(
        version="v4",
        expiration=timedelta(minutes=20),
        method="GET"
    )
    count = extract_frames(url, basename, pathname)

3. 移除本地服务账号密钥的冗余依赖 #

你当前用本地key.json生成签名URL，在Cloud Run里完全没必要——Cloud Run的服务账号已经拥有GCS访问权限，直接调用src_blob.generate_signed_url时不指定credentials，会自动使用应用默认凭据（ADC），既避免了读取本地文件的开销，又更安全（不需要把密钥打包到镜像里）。

4. 额外的GCS上传优化 #

• 启用分块上传：对于单帧JPG可能效果有限，但可以设置blob.chunk_size = 8 * 1024 * 1024（8MB分块），让大文件上传时自动并行分块；
• 避免频繁创建Blob对象：如果帧的路径结构固定，可以提前复用部分路径构造逻辑，但你这里每个帧名称都唯一，这个优化空间不大。

5. Cloud Run配置补充调整 #

虽然你说调过CPU没效果，但如果帧编码（cv2.imencode是CPU密集型）是隐性瓶颈，可以尝试把Cloud Run的CPU核数调到2核，内存对应升到1GB，编码速度上去了，上传队列的供给才不会断档。另外你的Cloud Run超时（15分钟）比PubSub的10分钟长，这个配置没问题，记得把PubSub的重试策略设为只重试失败任务，避免重复处理。

修复后可以加个日志统计，比如每上传1000帧打印一次耗时，看看是否能达到每秒几百帧的速度——GCS的写入限制远高于这个，正常情况下28000帧应该能在10分钟内完成。

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