asyncio run_in_executor上传Google Drive超时但完成的问题咨询

推测正确性分析 #

你的推测不准确。TimeoutError本质是线程池里执行的_perform_upload函数内部的网络请求超时，和「第一个线程等待第二个线程」没有关系。run_in_executor是把任务丢到线程池独立执行，线程之间不存在互相等待触发超时的逻辑。之所以看起来是第一个完成的文件报错，大概率是第一个文件的上传请求（尤其是可续传上传的后续分块请求）在网络层面触发了超时，但Google Drive的可续传机制让上传最终完成了。

解决超时问题的方案 #

1. 给Google Drive客户端设置自定义超时参数 #

直接在构建Drive服务时配置更长的超时时间，避免默认超时过短导致误报：

def build_g_drive_service():
    creds = authenticated_creds
    # 配置连接超时10秒，读取超时10分钟（可根据文件大小调整）
    session = requests.Session()
    session.timeout = (10, 600)
    drive = build(
        CONSTANTS.G_DRIVE,
        CONSTANTS.G_DRIVE_VERSION,
        credentials=creds,
        cache_discovery=False,
        http=creds.authorize(session)
    )
    return drive

2. 修复并发上传逻辑（当前代码仅上传单个文件） #

要实现真正的并发上传，需要创建多个run_in_executor任务并同时等待执行：

async def batch_upload(file_list):
    loop = asyncio.get_running_loop()
    # 根据Google Drive API配额限制，合理设置线程数
    with ThreadPoolExecutor(max_workers=2) as pool:
        # 批量生成上传协程
        upload_tasks = [
            loop.run_in_executor(pool, _perform_upload, meta, media)
            for meta, media in file_list
        ]
        # 并发执行，return_exceptions=True避免单个任务异常打断所有上传
        results = await asyncio.gather(*upload_tasks, return_exceptions=True)
    
    # 后续可遍历results处理成功/失败情况
    for idx, res in enumerate(results):
        if isinstance(res, Exception):
            print(f"第{idx+1}个文件上传出错: {res}")

# 调用示例：传入多个(文件元数据, 媒体对象)的元组
await batch_upload([(meta1, media1), (meta2, media2)])

3. 给上传函数添加超时重试机制 #

针对临时网络超时这类问题，添加指数退避重试：

from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests

@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避等待
    retry=retry_if_exception_type((requests.exceptions.Timeout,))  # 仅针对超时重试
)
def _perform_upload(file_metadata, media):
    drive = build_g_drive_serivce()
    drive.files().create(
        uploadType='resumable',
        body=file_metadata,
        media_body=media,
        supportsAllDrives=True
    ).execute()

更优的并发上传方式 #

如果涉及大量文件上传，推荐：

• 严格控制线程池max_workers数量，避免触发Google Drive的API速率限制
• 优先使用Google Drive官方支持的断点续传逻辑，大文件分块上传，出现异常时从断点继续
• 若追求纯异步无线程池开销，可基于aiohttp封装异步上传逻辑，替代线程池方案

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