如何在Python中使用标准boto3库并行下载多个S3对象？

当然可以！虽然boto3本身是同步库，但我们完全可以结合Python自带的并发工具来实现S3文件的并发下载，不用依赖aiobotocore。下面我给你两种常用的实现方案，都是基于标准库和boto3的，直接就能用：

S3下载属于IO密集型操作，线程切换的开销远低于进程，所以用线程池是最高效的选择。ThreadPoolExecutor是Python标准库提供的高级并发工具，能帮你轻松管理线程生命周期。

import boto3
from concurrent.futures import ThreadPoolExecutor

def download_s3_file(bucket_name, s3_key, local_file_path):
    # 每个线程创建S3客户端（也可以在主线程创建后传入，更高效）
    s3_client = boto3.client('s3')
    try:
        s3_client.download_file(bucket_name, s3_key, local_file_path)
        print(f"✅ 成功下载: {s3_key} -> {local_file_path}")
    except Exception as e:
        print(f"❌ 下载失败 {s3_key}: {str(e)}")

def main():
    # 替换成你的桶名和要下载的文件列表
    bucket_name = "your-target-bucket"
    files_to_download = [
        ("docs/report.pdf", "./local_report.pdf"),
        ("images/photo.png", "./local_photo.png"),
        ("data/dataset.csv", "./local_dataset.csv")
    ]
    
    # 并发数建议设为5-10，避免触发S3请求限制
    max_workers = 8
    
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 批量提交下载任务
        futures = [
            executor.submit(download_s3_file, bucket_name, s3_key, local_path)
            for s3_key, local_path in files_to_download
        ]
        
        # 等待所有任务完成（如果不需要同步等待可以省略这步）
        for future in futures:
            future.result()

if __name__ == "__main__":
    main()

优化小技巧： #

你可以在主线程提前创建一个S3客户端，然后传递给每个下载函数，这样能减少重复初始化客户端的开销——boto3的客户端是线程安全的，完全可以在多个线程间共享：

# 修改下载函数，接受客户端参数
def download_s3_file(s3_client, bucket_name, s3_key, local_file_path):
    try:
        s3_client.download_file(bucket_name, s3_key, local_file_path)
        print(f"✅ 成功下载: {s3_key} -> {local_file_path}")
    except Exception as e:
        print(f"❌ 下载失败 {s3_key}: {str(e)}")

# 在main里创建客户端
def main():
    s3_client = boto3.client('s3')
    # ...其他代码不变
    futures = [
        executor.submit(download_s3_file, s3_client, bucket_name, s3_key, local_path)
        for s3_key, local_path in files_to_download
    ]

如果你需要更精细地控制线程的创建和运行，可以直接用Python的threading模块手动管理线程：

import boto3
import threading

def download_s3_file(bucket_name, s3_key, local_file_path):
    s3_client = boto3.client('s3')
    try:
        s3_client.download_file(bucket_name, s3_key, local_file_path)
        print(f"✅ 成功下载: {s3_key} -> {local_file_path}")
    except Exception as e:
        print(f"❌ 下载失败 {s3_key}: {str(e)}")

def main():
    bucket_name = "your-target-bucket"
    files_to_download = [
        ("docs/report.pdf", "./local_report.pdf"),
        ("images/photo.png", "./local_photo.png")
    ]
    
    # 创建并启动线程
    threads = []
    for s3_key, local_path in files_to_download:
        thread = threading.Thread(
            target=download_s3_file,
            args=(bucket_name, s3_key, local_file_path)
        )
        threads.append(thread)
        thread.start()
    
    # 等待所有线程执行完成
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    main()

• S3请求限制：AWS S3对每个桶有默认的请求速率限制（比如前缀下每秒3500个写请求、5500个读请求），并发数不要设置过高，否则可能遇到RequestLimitExceeded错误。如果需要更高的并发，可以给文件设置不同的前缀（比如data/a/file.csv、data/b/file.csv）来分散请求。
• 异常处理：一定要加上异常捕获，避免单个文件下载失败导致整个程序崩溃。
• 大文件处理：如果下载大文件，可以用download_fileobj配合分块下载，并发思路和上面一致——每个线程处理一个大文件，或者拆分大文件为多个分块并发下载。

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