aioboto3批量上传本地文件到AWS S3出现多种错误问题求助

错误根因 #

你遇到的所有报错本质是无限制并发导致的资源过载：一次性启动所有上传任务，会同时发起大量TCP连接到S3 endpoint，触发本地DNS查询超限、端口占满、S3服务端主动断开空闲/超限连接，最终出现连接失败、超时错误。
aioboto3本身是完全适配该场景的方案，按以下步骤修改即可：

修复步骤 #

• 1. 限制并发上传数量
     通过asyncio.Semaphore控制同时上传的任务数，普通家用/办公网络环境建议初始设置为5~10，再根据实际上传速度调整。
• 2. 复用S3客户端
     不要为每个上传任务单独创建Session和S3客户端，复用全局客户端可以大幅减少连接建立开销，提升连接利用率。
• 3. 调整客户端默认配置
     增加连接超时时间、重试次数，调整连接池大小匹配并发数。
• 4. 配置大文件分片上传参数
     针对大文件开启自动分片，避免单请求超时。

修改后可运行代码示例 #

import asyncio
import aioboto3
from pathlib import Path
from botocore.config import Config

# 配置参数区
ACCESS_KEY = "你的AK"
SECRET_KEY = "你的SK"
bucket = "你的桶名"
bucket_folder = "桶内路径前缀/"
MAX_CONCURRENT_UPLOADS = 8 # 并发数可自行调整
# 自定义S3客户端配置
s3_config = Config(
    connect_timeout=30,
    read_timeout=120,
    retries={'max_attempts': 10, 'mode': 'standard'},
    max_pool_connections=MAX_CONCURRENT_UPLOADS
)

# 全局信号量控制并发
semaphore = asyncio.Semaphore(MAX_CONCURRENT_UPLOADS)

async def upload(staging_path, s3_client):
    async with semaphore: # 申请信号量，超过并发数的任务等待
        v_file_path = Path(staging_path)
        blob_s3_key = bucket_folder + staging_path.rsplit('\\', 1)[-1]
        try:
            with v_file_path.open("rb") as spfp:
                print(f"Uploading {blob_s3_key} to s3")
                await s3_client.upload_fileobj(
                    spfp, 
                    bucket, 
                    blob_s3_key,
                    Config={'multipart_threshold': 8*1024*1024, 'multipart_chunksize': 8*1024*1024} # 大于8MB自动分片
                )
                print(f"Finished Uploading {blob_s3_key} to s3")
        except Exception as e:
            print(f"Unable to s3 upload {staging_path} to {blob_s3_key}: {e} ({type(e)})")
            return ""

async def run_proc(file_upload_list):
    # 全局只创建一个S3客户端复用
    session = aioboto3.Session(aws_access_key_id=ACCESS_KEY, aws_secret_access_key=SECRET_KEY)
    async with session.client("s3", config=s3_config) as s3:
        tasks = [asyncio.create_task(upload(i, s3)) for i in file_upload_list]
        await asyncio.gather(*tasks)

# 调用示例
# if __name__ == "__main__":
#     file_list = ["你的文件路径1", "你的文件路径2"]
#     asyncio.run(run_proc(file_list))

额外调优建议 #

如果是Windows环境出现DNS解析失败，可以手动在系统hosts文件中添加S3 endpoint的IP映射，避免频繁DNS查询超时。
如果上传的都是小于1MB的小文件，可以适当把并发数上调到15~20，如果是大文件为主则适当降低并发数。

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