使用aiofiles&aiohttp加速文件上传：解决tracemalloc报错及方案咨询

一、当前代码的问题及修复 #

你的代码存在几个直接问题，是触发tracemalloc提示及上传失败的根源：

1. 协程未执行：session.post是异步方法，必须加await，否则协程不会真正运行，还会引发资源泄漏。
2. 参数冲突：session.post不能同时传data和json参数，aiohttp不允许这种请求构造方式。
3. 内存过载风险：await fp.read()会把整个文件读进内存，大文件会瞬间占用大量内存，触发内存分配警告（即你看到的tracemalloc提示）。

修复后的基础可运行代码 #

import asyncio
import aiofiles
import aiohttp
from aiohttp import FormData

async def upload_file(session, local_path):
    form = FormData()
    # 添加表单字段
    form.add_field('me', 'yo')
    # 流式上传，不一次性加载整个文件到内存
    async with aiofiles.open(local_path, 'rb') as fp:
        form.add_field('file', fp, filename=local_path.split('/')[-1])
        # 必须await异步请求
        async with session.post('http://my_url', data=form) as response:
            # 可选：校验响应状态
            if response.status == 200:
                print(f"{local_path} 上传成功")
            else:
                print(f"{local_path} 上传失败，状态码：{response.status}")

async def upload_files(paths):
    # 限制并发数，避免资源过载
    semaphore = asyncio.Semaphore(5)
    async def bounded_upload(path):
        async with semaphore:
            await upload_file(session, **path)
    
    async with aiohttp.ClientSession() as session:
        # return_exceptions=True 避免单个失败导致全部任务终止
        await asyncio.gather(*[bounded_upload(path) for path in paths], return_exceptions=True)

async def main():
    await upload_files([
        {'local_path': '1.txt'},
        # 可添加更多文件路径
    ])

if __name__ == "__main__":
    # 启用tracemalloc定位内存问题（调试用）
    import tracemalloc
    tracemalloc.start()
    
    asyncio.run(main())
    
    # 可选：打印内存分配快照，排查内存热点
    snapshot = tracemalloc.take_snapshot()
    top_stats = snapshot.statistics('lineno')
    print("\n[内存分配统计]")
    for stat in top_stats[:5]:
        print(stat)

二、tracemalloc提示的直接处理 #

如果需要精准定位内存分配的具体代码行，只需在代码开头添加以下代码即可开启追踪：

import tracemalloc
tracemalloc.start()

运行后会输出详细的内存分配回溯信息，帮你找到内存占用过高的代码位置。

三、更优的文件上传实现方案 #

1. 流式上传（核心优化） #

直接将异步文件对象传给FormData，aiohttp会自动流式上传，无需一次性加载整个文件到内存，大幅降低内存占用，尤其适合大文件场景。

2. 并发数控制 #

用asyncio.Semaphore限制同时上传的文件数量，避免本地网络/内存过载，也防止被服务器限流。

3. 错误隔离 #

在asyncio.gather中添加return_exceptions=True，单个文件上传失败不会导致整个任务组崩溃，后续可单独处理失败的任务。

4. 上传进度监控（可选） #

如果需要显示上传进度，可以包装文件对象统计已读取字节数：

class ProgressFileReader:
    def __init__(self, file_obj, total_size):
        self.file_obj = file_obj
        self.total_size = total_size
        self.read_size = 0

    async def read(self, n=-1):
        chunk = await self.file_obj.read(n)
        self.read_size += len(chunk)
        progress = (self.read_size / self.total_size) * 100
        print(f"上传进度：{progress:.2f}%")
        return chunk

# 使用方式替换原文件读取逻辑：
async with aiofiles.open(local_path, 'rb') as fp:
    file_size = (await fp.stat()).st_size
    progress_reader = ProgressFileReader(fp, file_size)
    form.add_field('file', progress_reader, filename=local_path.split('/')[-1])

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