如何使用ThreadPoolExecutor加快Google Drive数据传输速度

注意：Google API 生成的 service 实例非线程安全，禁止多线程共享同一个实例，否则会出现请求串号、连接报错等问题，每个工作线程必须独立初始化自己的service连接

改造后的完整代码如下：

from concurrent.futures import ThreadPoolExecutor, as_completed
from google.oauth2 import service_account
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
import pandas as pd

def upload_single_file(file_id: str, file_name: str, local_dir: str, sa_key_path: str, scopes: list):
    # 每个工作线程独立初始化凭证和service
    creds = service_account.Credentials.from_service_account_file(sa_key_path, scopes=scopes)
    service = build('drive', 'v3', credentials=creds, cache_discovery=False)
    # 小文件可以关闭resumable减少握手开销，大文件建议chunksize设为10*1024*1024（10MB）
    media = MediaFileUpload(
        local_dir + file_name, 
        resumable=True,
        chunksize=10*1024*1024
    )
    resp = service.files().update(fileId=file_id, media_body=media).execute()
    service.close()
    return resp

def update_tables(max_workers: int = 6):
    dir_path_1 = r'/home/.../'
    dir_path_2 = r'/home/.../test.json'
    SCOPES = ['https://www.googleapis.com/auth/drive']

    # 主线程仅拉取一次文件列表，不需要放到线程里
    creds = service_account.Credentials.from_service_account_file(dir_path_2, scopes=SCOPES)
    main_service = build('drive', 'v3', credentials=creds, cache_discovery=False)
    results = main_service.files().list(
        pageSize=100,
        fields="nextPageToken, files(id, name)",
        q="'1................P' in parents"
    ).execute()
    main_service.close()

    df = pd.DataFrame(results['files'])

    # 线程池并发执行上传
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        tasks = [
            executor.submit(upload_single_file, file_id, file_name, dir_path_1, dir_path_2, SCOPES)
            for file_id, file_name in zip(df['id'], df['name'])
        ]
        for task in as_completed(tasks):
            try:
                print(task.result())
            except Exception as e:
                print(f"文件上传异常：{str(e)}")

update_tables()

改造核心说明：

• 将单文件上传逻辑抽离为独立worker函数，每个线程自行初始化service连接，规避线程安全问题
• 初始化build时添加cache_discovery=False参数，跳过重复拉取API描述文档的步骤，减少连接初始化耗时
• 用as_completed按任务实际完成顺序处理结果，不会因为单个慢任务阻塞整体流程
• max_workers建议从4开始调试，最高不要超过10，避免触发Google Drive API的频率限制返回429错误

• 调整上传参数：小于10MB的小文件直接设置resumable=False走单次上传，去掉断点续传的握手开销；大文件将chunksize调整为10MB-25MB区间，平衡请求次数和单请求失败重传成本
• 配置自动重试：初始化service时添加num_retries=5参数，内置的退避重试逻辑会自动处理429限流、5xx服务端错误，不需要手动重跑任务
• 补全文件列表分页逻辑：当前代码pageSize为100，如果目标文件夹文件数超过100，需要循环读取nextPageToken拉取全量文件列表，避免漏传导致的二次补传耗时
• 优化网络链路：如果服务部署在国内，使用稳定的国际出口专线；如果部署在公有云，优先选择和Drive存储区域同地域的节点，走内网传输可以跑满带宽且无额外流量费
• 使用批量请求接口：大量小文件更新时可以用Drive API的Batch功能，将最多100个update请求打包为单个HTTP请求发送，减少TCP握手和请求头开销
• 精简返回字段：调用files().update()时指定fields='id,name'这类你实际需要的字段，不要返回全量文件元数据，减少响应体传输大小

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