使用asyncio+aiobotocore批量复制S3文件遇异步问题求助

核心问题原因 #

1. 异步效果未达预期：你在单个文件的复制循环中逐个await复制操作，导致事件循环被串行阻塞——必须等当前文件的所有复制完成，才会调度下一个文件的任务，完全没有利用asyncio的并发能力。
2. print无输出：使用end=''时，Python标准输出默认是行缓冲模式，没有换行符的情况下不会主动刷新缓冲区，导致内容暂存内存无法显示。

优化后的代码实现 #

基础并发版本 #

import asyncio
import uuid
import aiobotocore

async def async_duplicate_files_in_bucket(session, bucket_name, how_many_times):
    async with session.create_client('s3') as client:
        paginator = client.get_paginator('list_objects_v2')
        all_tasks = []
        
        async for page in paginator.paginate(Bucket=bucket_name):
            for obj in page.get('Contents', []):
                src_key = obj['Key']
                # 为当前文件创建所有复制任务，暂不执行
                file_copy_tasks = [
                    client.copy_object(
                        Bucket=bucket_name,
                        CopySource={'Bucket': bucket_name, 'Key': src_key},
                        Key=f"{src_key}.{uuid.uuid4()}"
                    )
                    for _ in range(how_many_times)
                ]
                # 批量添加当前文件的复制任务组
                all_tasks.extend(file_copy_tasks)
                # 刷新输出缓冲区，确保实时显示
                print('-' * how_many_times, end='', flush=True)
        
        # 并发执行所有复制任务
        await asyncio.gather(*all_tasks)

带并发数控制的版本（避免S3限流） #

如果文件数量或复制次数极大，直接并发所有任务可能触发S3的请求限流，建议用信号量控制并发数：

import asyncio
import uuid
import aiobotocore

async def async_duplicate_files_in_bucket(session, bucket_name, how_many_times, max_concurrent=50):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def copy_single(src_key):
        async with semaphore:
            dest_key = f"{src_key}.{uuid.uuid4()}"
            await client.copy_object(
                Bucket=bucket_name,
                CopySource={'Bucket': bucket_name, 'Key': src_key},
                Key=dest_key
            )
            print('-', end='', flush=True)
    
    async with session.create_client('s3') as client:
        paginator = client.get_paginator('list_objects_v2')
        all_tasks = []
        
        async for page in paginator.paginate(Bucket=bucket_name):
            for obj in page.get('Contents', []):
                src_key = obj['Key']
                # 批量添加当前文件的复制任务
                all_tasks.extend([copy_single(src_key) for _ in range(how_many_times)])
        
        await asyncio.gather(*all_tasks)

asyncio关键使用指导 #

• 批量调度协程：不要在循环中逐个await协程，而是将所有协程对象收集到列表中，用asyncio.gather(*tasks)一次性并发执行，让事件循环可以同时调度多个任务。
• 避免阻塞操作：异步函数中绝对不能调用同步阻塞的方法（比如time.sleep、同步版boto3接口），必须使用对应的异步实现（如asyncio.sleep、aiobotocore接口）。
• 输出缓冲处理：当需要实时显示无换行的输出时，必须给print加上flush=True参数，强制刷新输出缓冲区。
• 并发数控制：面对外部服务（如S3）时，用asyncio.Semaphore限制同时执行的任务数量，避免因请求过载被服务端限流或拒绝。

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