如何使用aioboto3并发列举S3桶对象并提取文件对应的唯一父目录路径

核心问题原因 #

1. 原有代码使用已被官方不推荐的list_objects接口，分页逻辑存在兼容性缺陷，大概率是你拿不到全量数据的核心原因。
2. aioboto3的异步paginator仅支持异步等待单页响应，不会并发拉取多个分页，所以整体执行逻辑还是串行，速度慢符合预期。
3. 循环内调用print属于高频IO操作，会大幅拖慢处理速度。
4. 拉取全量对象元数据再提取目录的逻辑，在桶内文件数量远大于目录数量时会做大量无用IO，性能极低。

问题1：加快查找与保存目录速度的方案 #

方案1（推荐）：使用Delimiter参数直接查询目录 #

不用拉取全量文件元数据，S3接口会直接返回目录前缀，性能提升十倍到百倍不等，适配目录深度不固定的场景：

import asyncio
from typing import Set

async def get_all_folders(bucket: str, prefix: str) -> Set[str]:
    subfolders = set()
    prefix_queue = asyncio.Queue()
    await prefix_queue.put(prefix)

    while not prefix_queue.empty():
        current_prefix = await prefix_queue.get()
        paginator = self.s3_client.get_paginator("list_objects_v2")
        async for result in paginator.paginate(
            Bucket=bucket,
            Prefix=current_prefix,
            Delimiter='/'
        ):
            # 直接获取当前层级的所有目录
            for common_prefix in result.get("CommonPrefixes", []):
                folder_path = common_prefix.get("Prefix")
                subfolders.add(folder_path)
                # 下一级目录入队继续递归
                await prefix_queue.put(folder_path)
    return subfolders

方案2：优化全量拉取逻辑 #

如果必须拉取全量文件确认目录（比如存在单文件无上级目录占位的特殊场景），按如下逻辑优化：

• 替换为list_objects_v2接口，设置单页最大拉取数量1000
• 移除循环内高频print，仅分页完成后打日志
• 替换os.path.dirname为字符串分割，减少不必要的系统调用

subfolders = set()
paginator = self.s3_client.get_paginator("list_objects_v2")

async for result in paginator.paginate(
    Bucket=bucket,
    Prefix=prefix,
    PaginationConfig={'PageSize': 1000}
):
    for file in result.get("Contents", []):
        key = file.get("Key")
        # 跳过目录占位文件
        if key.endswith('/'):
            continue
        # 直接分割字符串获取父目录
        current_path = key.rsplit('/', 1)[0] + '/'
        subfolders.add(current_path)
    print("单分页处理完成")

return subfolders

进阶提速方案 #

将扫描前缀拆分为多个独立分片，比如prefix/a/、prefix/b/，同时启动多个异步任务并行跑上述逻辑，可充分利用带宽，速度提升数倍到数十倍。

问题2：全量分页获取方案 #

aioboto3的list_objects_v2接口paginator已经封装了自动分页逻辑，正常会迭代所有分页返回全量数据。如果仍然出现数据缺失，可手动判断分页状态拉取下一页：

continuation_token = None
subfolders = set()

while True:
    kwargs = {
        "Bucket": bucket,
        "Prefix": prefix,
        "MaxKeys": 1000
    }
    if continuation_token:
        kwargs["ContinuationToken"] = continuation_token
    
    resp = await self.s3_client.list_objects_v2(**kwargs)
    for file in resp.get("Contents", []):
        key = file.get("Key")
        if key.endswith('/'):
            continue
        current_path = key.rsplit('/', 1)[0] + '/'
        subfolders.add(current_path)
    
    # 判断是否还有下一页
    if not resp.get("IsTruncated"):
        break
    continuation_token = resp.get("NextContinuationToken")

return subfolders

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