如何优化基于Python的S3异步大视频上传性能？

1. 你的代码是异步实现，但存在核心性能瓶颈 #

你的代码确实基于asyncio、httpx和aioboto3搭建了异步框架，但单视频文件的分片上传是串行执行的：在async_upload中，你通过async for chunk in stream逐个读取分片，再逐个await s3.upload_part，每个分片必须等前一个上传完成才会启动下一个。这种设计完全没发挥异步并行上传分片的优势，导致单文件上传耗时过长，即便多文件并行，整体效率也无法提升。

2. 性能优化核心思路 #

关键改进：并行上传单文件的多个分片 #

S3分片上传原生支持同时上传多个分片，我们可以利用asyncio.gather并行处理分片的读取与上传，同时用信号量控制并发数，避免触发S3请求频率限制或耗尽本地资源。

辅助优化点 #

• 复用aioboto3.Session和S3客户端，避免每次上传重复创建连接
• 调整分片大小（建议10-50MB，更小的分片可提升并行度，但会增加请求次数，按需平衡）
• 延长网络请求超时时间，适配大分片的传输需求
• 给每个视频分配唯一文件名，避免S3上的文件覆盖

3. 修改后的代码示例 #

import asyncio
import httpx
from botocore.exceptions import ClientError
import aioboto3
from typing import AsyncIterator, List, Tuple


# 全局复用session，减少连接创建开销
session = aioboto3.Session()

async def upload_single_part(s3_client, bucket: str, key: str, upload_id: str, part_num: int, chunk: bytes) -> Tuple[int, str]:
    """单独上传一个分片，返回分片号和对应的ETag"""
    resp = await s3_client.upload_part(
        Bucket=bucket,
        Key=key,
        PartNumber=part_num,
        UploadId=upload_id,
        Body=chunk
    )
    print(f'Chunk {part_num} uploaded')
    return (part_num, resp['ETag'])

async def async_upload(stream: AsyncIterator[bytes], 
                        destination_key: str,
                        max_concurrent_parts: int = 5
                        ) -> None:
    """
    异步并行上传视频文件分片
    Args:
        stream: 视频流迭代器
        destination_key: S3存储键
        max_concurrent_parts: 最大并行分片数，按需调整
    """
    async with session.client('s3') as s3:
        resp = await s3.create_multipart_upload(Bucket=bucket_name, Key=destination_key)
        upload_id = resp['UploadId']
        semaphore = asyncio.Semaphore(max_concurrent_parts)

        async def bounded_upload(part_num: int, chunk: bytes):
            """用信号量控制并发数，避免请求过载"""
            async with semaphore:
                return await upload_single_part(s3, bucket_name, destination_key, upload_id, part_num, chunk)

        try:
            upload_tasks = []
            part_num = 1
            # 批量收集分片上传任务
            async for chunk in stream:
                task = bounded_upload(part_num, chunk)
                upload_tasks.append(task)
                part_num += 1
            
            # 等待所有分片上传完成
            parts = await asyncio.gather(*upload_tasks)
            
            # S3要求分片列表必须按PartNumber升序排列
            sorted_parts = sorted(parts, key=lambda x: x[0])
            formatted_parts = [{'ETag': etag, 'PartNumber': num} for num, etag in sorted_parts]

            await s3.complete_multipart_upload(
                Bucket=bucket_name,
                Key=destination_key,
                UploadId=upload_id,
                MultipartUpload={'Parts': formatted_parts}
            )
            print(f'Upload completed: {destination_key}')
        except Exception as e:
            await s3.abort_multipart_upload(
                Bucket=bucket_name,
                Key=destination_key,
                UploadId=upload_id
            )
            print(f'Upload failed for {destination_key}: {str(e)}')


async def upload_video(url: str, file_idx: int):
    # 生成唯一文件名，避免S3文件覆盖
    unique_file_name = f'Video_{file_idx}.MP4'
    s3_key = f'{folder_name}/{unique_file_name}'
    
    async with httpx.AsyncClient(timeout=httpx.Timeout(20, read=30)) as request:
        async with request.stream("GET", url) as response:
            if response.status_code == 200:
                try:
                    # 调整分片大小为20MB，适配并行上传
                    await async_upload(response.aiter_bytes(chunk_size=20*1024*1024), s3_key)
                except ClientError as e:
                    print(f'Upload failed for {s3_key}: {str(e)}')
            else:
                print(f'Bad response {response.status_code} for URL: {url}')

async def main():
    urls = [
        'https://api.example.com/somevideo.mp4',
        'https://api.example.com/somevideo.mp4',
        'https://api.example.com/somevideo.mp4'
    ]
    
    tasks = [
        upload_video(url, idx)
        for idx, url in enumerate(urls)
    ]
   
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    bucket_name = 'Test-Bucket'
    folder_name = 'Test'

    asyncio.run(main())

4. 注意事项 #

• 并发数可根据你的网络带宽和S3配额调整，默认5个并发较为稳妥，过高可能触发请求限制
• 若允许先将视频下载到本地，可进一步优化为读取本地文件分片并行上传，效率会更稳定
• 确保你的AWS密钥权限包含s3:CreateMultipartUpload、s3:UploadPart、s3:CompleteMultipartUpload和s3:AbortMultipartUpload权限

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