Auto算力云上传文件夹的高效解决方案：并发处理与断点续传实践

1次阅读

没有评论

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

在 Auto 算力云平台上进行大规模文件夹上传时，开发者经常会遇到以下几个典型问题：

网络抖动导致传输中断 ：大文件上传过程中，网络不稳定可能导致连接断开，需要重新上传整个文件，浪费时间和资源。
海量小文件处理效率低 ：当文件夹中包含大量小文件时，传统的单线程上传方式效率极低，上传速度成为瓶颈。
任务管理复杂 ：上传过程中如果任务失败，缺乏有效的断点续传机制，导致重复劳动和进度丢失。

这些问题在大规模数据迁移（如 TB 级数据）时尤为突出，亟需一种高效、可靠的解决方案。

针对文件夹上传，常见的解决方案有以下几种：

HTTP 直传 ：简单易用，但无法处理大文件或网络不稳定的情况，缺乏断点续传能力。
分片上传 ：将文件分割为多个小块并行上传，支持断点续传，但实现复杂度较高。
SDK 封装 ：使用云平台提供的 SDK，简化开发流程，但灵活性较低，可能无法满足定制化需求。

综合来看，分片上传是最适合大规模文件夹上传的方案，能够显著提升上传速度和可靠性。

我们采用以下架构实现高效上传：

文件扫描与分片 ：遍历文件夹中的所有文件，并为每个文件生成分片任务。
任务队列 ：将分片任务放入队列，由多个工作线程并发处理。
断点续传 ：记录已上传的分片信息，支持从断点继续上传。
进度监控 ：实时统计上传进度，提供回调接口供用户监控。

以下是一个分片上传的代码示例，包含分片算法、错误重试逻辑和进度回调实现：

import os
import threading
from queue import Queue

# 分片大小动态调整（单位：字节）DEFAULT_CHUNK_SIZE = 4 * 1024 * 1024  # 4MB

class UploadManager:
    def __init__(self, folder_path, max_workers=4):
        self.folder_path = folder_path
        self.max_workers = max_workers
        self.task_queue = Queue()
        self.progress = {}

    def scan_files(self):
        # 遍历文件夹，生成分片任务
        for root, _, files in os.walk(self.folder_path):
            for file in files:
                file_path = os.path.join(root, file)
                self._split_file(file_path)

    def _split_file(self, file_path):
        file_size = os.path.getsize(file_path)
        chunk_size = self._calculate_chunk_size(file_size)
        chunks = (file_size + chunk_size - 1) // chunk_size

        for i in range(chunks):
            start = i * chunk_size
            end = min((i + 1) * chunk_size, file_size)
            self.task_queue.put((file_path, i, start, end))

    def _calculate_chunk_size(self, file_size):
        # 根据文件大小动态调整分片大小
        if file_size < 10 * 1024 * 1024:  # 小于 10MB 的文件不分片
            return file_size
        return DEFAULT_CHUNK_SIZE

    def upload_chunk(self, file_path, chunk_id, start, end):
        # 实现分片上传逻辑
        retry_count = 0
        max_retries = 3

        while retry_count < max_retries:
            try:
                # 模拟上传逻辑
                print(f"Uploading {file_path} chunk {chunk_id} ({start}-{end})")
                # 上传成功后更新进度
                self.progress[file_path] = self.progress.get(file_path, 0) + (end - start)
                break
            except Exception as e:
                retry_count += 1
                if retry_count >= max_retries:
                    print(f"Failed to upload {file_path} chunk {chunk_id}: {e}")
                else:
                    # 指数退避重试
                    time.sleep(2 ** retry_count)

    def start_upload(self):
        # 启动工作线程
        threads = []
        for _ in range(self.max_workers):
            t = threading.Thread(target=self._worker)
            t.start()
            threads.append(t)

        # 等待所有任务完成
        self.task_queue.join()
        for t in threads:
            t.join()

    def _worker(self):
        while True:
            task = self.task_queue.get()
            if task is None:
                break
            self.upload_chunk(*task)
            self.task_queue.task_done()

# 使用示例
manager = UploadManager("/path/to/folder")
manager.scan_files()
manager.start_upload()

为了避免内存溢出（OOM），需要注意以下几点：