auto算力云上传文件夹的架构设计与性能优化实践

1次阅读

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

在分布式计算场景中，auto 算力云需要频繁处理大规模文件夹上传任务。传统 FTP/HTTP 上传方式在此类场景下暴露明显缺陷：

网络闪断导致整个文件重传，浪费带宽资源
海量小文件传输时连接建立开销占比过高
缺乏完整性校验机制，可能引发数据不一致
单线程上传无法充分利用现代多核 CPU 性能

实测数据显示：当网络延迟超过 200ms 时，传统方式上传 500MB 文件夹的失败率高达 32%。

维度	分片上传	流式上传
网络容错	支持断点续传	失败需全量重传
内存占用	固定分片缓冲区	需维持长连接缓冲区
并行能力	天然支持多线程	单连接难以并行化
校验粒度	分片级校验	只能整体校验

flowchart TD
    A[客户端] -->| 分片预处理 | B[计算 MD5]
    B --> C[生成分片清单]
    C --> D[并发上传分片]
    D --> E[服务端]
    E -->| 并行校验 | F[校验分片 CRC32]
    F --> G[合并存储]
    G --> H[返回 ETag]

def chunk_file(file_path: str, chunk_size: int = 4 * 1024 * 1024) -> list:
    """
    动态分片算法：根据文件大小自动调整分片策略
    :param file_path: 源文件路径
    :param chunk_size: 基础分片大小（默认 4MB）:return: 分片信息列表
    """
    file_size = os.path.getsize(file_path)
    # 大文件采用固定分片，小文件动态调整
    adaptive_size = max(chunk_size, file_size // 100) if file_size < 1GB else chunk_size

    chunks = []
    with open(file_path, 'rb') as f:
        for i in itertools.count():
            offset = i * adaptive_size
            if offset >= file_size:
                break
            # 计算分片哈希
            f.seek(offset)
            data = f.read(adaptive_size)
            md5 = hashlib.md5(data).hexdigest()
            chunks.append({
                'index': i,
                'offset': offset,
                'size': len(data),
                'md5': md5
            })
    return chunks

class UploadStateMachine:
    def __init__(self, task_id: str):
        self.state = {
            'task_id': task_id,
            'completed': set(),  # 已上传分片索引
            'failed': defaultdict(int)  # 失败计数
        }

    def mark_complete(self, chunk_index: int):
        self.state['completed'].add(chunk_index)

    def mark_failed(self, chunk_index: int):
        self.state['failed'][chunk_index] += 1

    def should_retry(self, chunk_index: int) -> bool:
        return self.state['failed'].get(chunk_index, 0) < 3