FastAPI大文件快速传输最佳实践：200MB TIFF图像提速方案咨询

我需要将200MB大小的TIFF图像传输至FastAPI后端，常规文件上传方式耗时达5秒。请问实现大文件快速传输有哪些最佳实践？

我尝试了以下方法来提升传输速度，该方法在处理1.5MB的TIFF小文件时，将耗时从200ms缩短至100ms，但仅适用于小文件，无法适配200MB大文件：

我尝试的实现代码 #

服务端代码 #

import time
import aiofiles
import numpy as np
import cv2
from fastapi import FastAPI, Request, Response

app = FastAPI()

def add_text(img, text):
    # 自定义添加文字逻辑
    return img

@app.post('/upload_tiff_V2')
async def upload_tiff_V2(request: Request):
    start = time.time()
    body = b''
    try:
        filename = request.headers['filename']
        async with aiofiles.open(filename, 'wb') as f:
            async for chunk in request.stream():
                await f.write(chunk)
                body += chunk
        
        arr = np.frombuffer(body, dtype=np.uint8)
        img = cv2.imdecode(arr, cv2.IMREAD_COLOR)

        arr = add_text(img, "AVERNA")

        success, im = cv2.imencode('.tif', arr)
        headers = {'Content-Disposition': 'inline; filename="test.tif"'}
        end = time.time()
        print("TIFF处理执行时间：", (end-start) * 10**3, "ms")
    except Exception:
        return {"message": "文件上传出错"}
    return Response(im.tobytes(), headers=headers, media_type='image/tif')

客户端代码 #

import requests
import time

#############################################
## 上传大TIFF文件
#############################################
with open("big_image.tif", "rb") as f:
    data = f.read()
   
url = 'http://192.168.199.130:2123/upload_tiff_V2'
headers = {'filename': 'big_image.tif'}

start = time.time()
resp = requests.post(url=url, data=data, headers=headers)
end = time.time()

print("200MB TIFF文件上传执行时间：", (end-start) * 10**3, "ms")

大文件快速传输的最佳实践 #

1. 分块上传（断点续传） #

将大文件分割成10MB左右的小分片，客户端逐个发送分片，服务端接收后拼接。这种方式既能降低单次传输压力，也能在传输中断时仅重传失败分片，避免从头开始。

• 服务端：设计接口接收分片，记录已上传分片索引，所有分片上传完成后合并文件。
• 客户端：按固定大小分割文件，可选择并发/顺序发送分片，携带分片编号、总片数、文件唯一标识（如MD5）。

2. 启用HTTP/2或HTTP/3 #

HTTP/2支持多路复用，能在单个连接上同时传输多个请求/响应，减少TCP握手开销，大幅提升大文件传输效率。FastAPI基于Starlette，可通过Uvicorn启用HTTP/2：

uvicorn main:app --port 2123 --http h2

客户端使用支持HTTP/2的库（如httpx）即可适配。

3. 优化内存与IO操作 #

当前代码将整个文件读入内存（body += chunk），200MB文件会占用大量内存引发阻塞，优化方向：

• 服务端：不要缓存整个文件到内存，直接流式写入磁盘，处理时再从磁盘读取。
• 客户端：避免一次性读取整个文件，直接传入文件对象让requests自动分块发送：

with open("big_image.tif", "rb") as f:
    resp = requests.post(url=url, data=f, headers=headers)

4. 启用压缩传输 #

如果TIFF未压缩，传输前用zlib等工具压缩，服务端接收后解压。注意：若TIFF本身已是压缩格式（如LZW），压缩收益极低。

• 客户端：压缩数据后在Header中标识Content-Encoding: gzip。
• 服务端：根据Header解压数据后再处理。

5. 网络层面优化 #

• 增大TCP窗口大小：调整系统TCP参数（如Linux下net.ipv4.tcp_window_scaling），提升带宽利用率。
• 就近部署：跨地域场景使用CDN或边缘节点，减少网络延迟。

6. 异步并发传输 #

用aiohttp等异步库并发发送分片，利用IO等待时间发送更多数据，提升整体速度：

import aiohttp
import asyncio
import os

async def upload_chunk(session, url, chunk, chunk_idx, total_chunks, file_id):
    headers = {
        "filename": "big_image.tif",
        "file-id": file_id,
        "chunk-idx": str(chunk_idx),
        "total-chunks": str(total_chunks)
    }
    async with session.post(url, data=chunk, headers=headers) as resp:
        return await resp.json()

async def main():
    chunk_size = 10 * 1024 * 1024  # 10MB分片
    file_path = "big_image.tif"
    file_id = f"{os.path.getsize(file_path)}_{os.path.basename(file_path)}"
    total_chunks = 0
    chunks = []
    
    with open(file_path, "rb") as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            chunks.append(chunk)
            total_chunks += 1
    
    url = 'http://192.168.199.130:2123/upload_tiff_chunk'
    async with aiohttp.ClientSession() as session:
        tasks = [upload_chunk(session, url, chunk, i, total_chunks, file_id) for i, chunk in enumerate(chunks)]
        await asyncio.gather(*tasks)

asyncio.run(main())

对应的服务端分块处理示例：

from fastapi import FastAPI, Request, Header, Response
import aiofiles
import os
import numpy as np
import cv2

app = FastAPI()
UPLOAD_DIR = "./uploads"
os.makedirs(UPLOAD_DIR, exist_ok=True)

def add_text(img, text):
    # 自定义添加文字逻辑
    return img

@app.post('/upload_tiff_chunk')
async def upload_tiff_chunk(
    request: Request,
    file_id: str = Header(...),
    chunk_idx: int = Header(...),
    total_chunks: int = Header(...),
    filename: str = Header(...)
):
    chunk_path = os.path.join(UPLOAD_DIR, f"{file_id}_chunk_{chunk_idx}")
    async with aiofiles.open(chunk_path, 'wb') as f:
        async for chunk in request.stream():
            await f.write(chunk)
    
    # 检查所有分片是否上传完成
    all_chunks_exist = all(
        os.path.exists(os.path.join(UPLOAD_DIR, f"{file_id}_chunk_{i}")) 
        for i in range(total_chunks)
    )
    
    if all_chunks_exist:
        # 合并分片
        final_path = os.path.join(UPLOAD_DIR, filename)
        async with aiofiles.open(final_path, 'wb') as f:
            for i in range(total_chunks):
                chunk_path = os.path.join(UPLOAD_DIR, f"{file_id}_chunk_{i}")
                async with aiofiles.open(chunk_path, 'rb') as cf:
                    await f.write(await cf.read())
                os.remove(chunk_path)
        
        # 处理合并后的文件
        async with aiofiles.open(final_path, 'rb') as f:
            arr = np.frombuffer(await f.read(), dtype=np.uint8)
            img = cv2.imdecode(arr, cv2.IMREAD_COLOR)
            img = add_text(img, "AVERNA")
            success, im = cv2.imencode('.tif', img)
        
        headers = {'Content-Disposition': 'inline; filename="test.tif"'}
        return Response(im.tobytes(), headers=headers, media_type='image/tif')
    else:
        return {"status": "partial", "received": chunk_idx + 1, "total": total_chunks}

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