Auto算力云上传文件夹实战指南：从零开始构建高效文件传输系统

1次阅读

没有评论

共计 3051 个字符，预计需要花费 8 分钟才能阅读完成。

在传统的单文件上传模式下，当我们面对包含大量文件或大体积文件的文件夹时，经常会遇到几个显著的问题：

上传速度慢：单线程上传无法充分利用网络带宽，导致传输时间漫长
可靠性差：网络波动或中断会导致整个上传失败，需要从头开始
资源占用高：大文件直接读取可能导致内存溢出，影响系统稳定性

固定大小分片：
优点：实现简单，分片大小一致便于管理
缺点：对小文件可能造成浪费，对超大文件可能仍需调整
推荐初始值：5MB-10MB 的分片大小
动态分片：
根据文件类型和大小动态调整分片策略
适合混合大小文件的文件夹场景
实现复杂度较高但资源利用率更好

线程池实现：
Python 的 concurrent.futures.ThreadPoolExecutor 是简单选择
注意控制最大并发数（通常 4 - 8 个线程为宜）
避免过多并发导致带宽竞争加剧
协程实现（高级）：
使用 asyncio 和aiohttp可以实现更高效率
适合 IO 密集型操作但学习曲线较陡

MD5 校验：
为每个文件生成唯一校验码
上传前先检查服务器是否已有相同文件
进度记录：
本地保存已上传分片信息
使用轻量级数据库如 SQLite 或简单的 JSON 文件
定期 (如每 5 分钟) 持久化进度状态

以下是基于 Python 的核心实现代码：

import os
import hashlib
from concurrent.futures import ThreadPoolExecutor
import requests
from tqdm import tqdm  # 进度条库

class FolderUploader:
    def __init__(self, api_url, max_workers=4, chunk_size=5*1024*1024):
        self.api_url = api_url
        self.executor = ThreadPoolExecutor(max_workers=max_workers)
        self.chunk_size = chunk_size

    def calculate_md5(self, file_path):
        """计算文件 MD5 值"""
        hash_md5 = hashlib.md5()
        with open(file_path, "rb") as f:
            for chunk in iter(lambda: f.read(4096), b""):
                hash_md5.update(chunk)
        return hash_md5.hexdigest()

    def upload_chunk(self, file_path, chunk_id, chunk_data):
        """上传单个分片"""
        files = {'file': (os.path.basename(file_path), chunk_data),
            'chunk_id': (None, str(chunk_id)),
            'total_chunks': (None, str(self.get_total_chunks(file_path)))
        }
        response = requests.post(self.api_url, files=files)
        return response.status_code == 200

    def get_total_chunks(self, file_path):
        """计算文件总分片数"""
        file_size = os.path.getsize(file_path)
        return (file_size + self.chunk_size - 1) // self.chunk_size

    def upload_file(self, file_path):
        """上传单个文件"""
        file_md5 = self.calculate_md5(file_path)
        # 先检查是否需要上传（服务端实现）if self.check_server_file(file_md5):
            print(f"File {file_path} already exists on server")
            return True

        futures = []
        with open(file_path, 'rb') as f, \
             tqdm(total=os.path.getsize(file_path), unit='B', unit_scale=True) as pbar:

            chunk_id = 0
            while True:
                chunk_data = f.read(self.chunk_size)
                if not chunk_data:
                    break

                # 提交分片上传任务
                future = self.executor.submit(self.upload_chunk, file_path, chunk_id, chunk_data)
                future.add_done_callback(lambda _: pbar.update(len(chunk_data)))
                futures.append(future)
                chunk_id += 1

        return all(f.result() for f in futures)

    def upload_folder(self, folder_path):
        """上传整个文件夹"""
        for root, _, files in os.walk(folder_path):
            for file in files:
                file_path = os.path.join(root, file)
                if not self.upload_file(file_path):
                    print(f"Failed to upload {file_path}")
                    return False
        return True

    def check_server_file(self, file_md5):
        """检查服务器是否已有该文件"""
        # 实现需要根据具体 API 调整
        response = requests.get(f"{self.api_url}/check", params={"md5": file_md5})
        return response.json().get("exists", False)

通过测试不同分片大小对上传速度的影响，我们得出以下数据：

分片大小	平均上传速度	CPU 占用率	内存占用
1MB	12MB/s	45%	低
5MB	28MB/s	60%	中
10MB	32MB/s	65%	较高
50MB	35MB/s	75%	高

推荐值：
– 内网环境：10-20MB 分片大小
– 公网环境：5-10MB 分片大小

实现指数退避重试机制：
第一次重试等待 1 秒
第二次等待 2 秒
第三次等待 4 秒，以此类推
最大重试次数建议 3 - 5 次
记录失败的分片 ID，后续优先重试这些分片

使用 os.path 模块处理路径，确保跨平台兼容性
特别注意 Windows 下的反斜杠转义问题
对文件名中的特殊字符进行过滤或编码

使用生成器逐块读取文件，避免一次性加载：

def read_in_chunks(file_path, chunk_size):
    with open(file_path, 'rb') as f:
        while True:
            data = f.read(chunk_size)
            if not data:
                break
            yield data