深度优化OSS上传性能：多线程分片上传 vs 断点续传实战对比

def single_thread_upload(file, endpoint):
    start = time.time()
    connection = create_connection(endpoint)  # 建立连接耗时 T_connect
    upload_data(connection, file)             # 数据传输耗时 T_transfer
    connection.close()
    return time.time() - start

// 分片上传核心逻辑
public class MultipartUploader {
    private static final int PART_SIZE = 5 * 1024 * 1024; // 5MB分片
    
    public void upload(File file, String bucketName) {
        // 初始化分片上传
        InitiateMultipartUploadRequest initRequest = new InitiateMultipartUploadRequest(bucketName, file.getName());
        InitiateMultipartUploadResult initResponse = ossClient.initiateMultipartUpload(initRequest);
        
        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
        List<Future<PartETag>> futures = new ArrayList<>();
        
        // 分片并提交任务
        long fileLength = file.length();
        int partCount = (int) (fileLength / PART_SIZE);
        if (fileLength % PART_SIZE != 0) partCount++;
        
        for (int i = 0; i < partCount; i++) {
            long startPos = i * PART_SIZE;
            long curPartSize = Math.min(PART_SIZE, fileLength - startPos);
            UploadPartTask task = new UploadPartTask(initResponse.getUploadId(), 
                                                    bucketName, 
                                                    file.getName(), 
                                                    file, 
                                                    startPos, 
                                                    curPartSize, 
                                                    i + 1);
            futures.add(executor.submit(task));
        }
        
        // 等待所有分片完成
        List<PartETag> partETags = new ArrayList<>();
        for (Future<PartETag> future : futures) {
            partETags.add(future.get());
        }
        
        // 合并分片
        CompleteMultipartUploadRequest compRequest = new CompleteMultipartUploadRequest(
            bucketName, file.getName(), initResponse.getUploadId(), partETags);
        ossClient.completeMultipartUpload(compRequest);
    }
}

// 分片上传任务
class UploadPartTask implements Callable<PartETag> {
    // 实现分片上传细节
    @Override
    public PartETag call() throws Exception {
        // 读取文件分片
        // 创建UploadPartRequest
        // 执行分片上传
        // 返回PartETag
    }
}

def calculate_part_size(file_size):
    # 根据文件大小动态调整分片
    if file_size <= 50 * 1024 * 1024:   # <50MB
        return 1 * 1024 * 1024          # 1MB分片
    elif file_size <= 5 * 1024 * 1024 * 1024: # <5GB
        return 5 * 1024 * 1024          # 5MB分片
    else:
        return 10 * 1024 * 1024         # 10MB分片

// 基于带宽的动态线程池
int maxThreads = (int) (NetworkMonitor.getAvailableBandwidth() / (PART_SIZE / 1024.0));
executor = new ThreadPoolExecutor(
    corePoolSize, 
    maxThreads, 
    60L, TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(1000),
    new ThreadPoolExecutor.CallerRunsPolicy());

// 断点续传管理器
type ResumeUploader struct {
    uploadID    string
    partTracker *PartTracker // 分片状态跟踪器
}

func (u *ResumeUploader) Upload(file *os.File) error {
    // 尝试加载进度
    if err := u.loadProgress(); err != nil {
        // 初始化上传
        u.initUpload()
    }
    
    // 获取待上传分片
    parts := u.partTracker.GetPendingParts()
    
    var wg sync.WaitGroup
    for _, part := range parts {
        wg.Add(1)
        go func(p Part) {
            defer wg.Done()
            // 上传分片
            etag := u.uploadPart(file, p)
            // 更新进度
            u.partTracker.CompletePart(p.Number, etag)
            u.saveProgress()
        }(part)
    }
    wg.Wait()
    
    // 完成上传
    return u.completeUpload()
}

// 分片状态跟踪
type PartTracker struct {
    parts map[int]PartStatus // 分片号->状态
}

type PartStatus struct {
    Start    int64
    End      int64
    ETag     string
    Complete bool
}

def save_upload_progress(upload_id, part_num, etag):
    # 高频小分片：每完成5个分片保存一次
    # 低频大分片：每个分片完成后立即保存
    # 超时分片：每30秒强制保存
    
    if part_num % 5 == 0 or part_size > 10*1024*1024:
        persist_to_db(upload_id, part_num, etag)
    else:
        cache_in_memory(upload_id, part_num, etag)

// 恢复上传时校验分片完整性
public boolean verifyPart(String uploadId, int partNumber, String expectedEtag) {
    ListPartsRequest listPartsRequest = new ListPartsRequest(bucket, key, uploadId);
    PartListing partListing = ossClient.listParts(listPartsRequest);
    
    for (PartSummary part : partListing.getParts()) {
        if (part.getPartNumber() == partNumber) {
            return part.getETag().equals(expectedEtag);
        }
    }
    return false;
}

class HybridUploader {
    private uploadId: string;
    private partTracker: PartTracker;
    private pauseSignal = false;
    
    async startUpload(file: File) {
        // 初始化或恢复上传
        if (!this.uploadId) {
            this.uploadId = await this.initOSSMultipartUpload();
        }
        
        // 加载或初始化分片状态
        this.partTracker = await PartTracker.load(file, this.uploadId) || 
                          new PartTracker(file, this.uploadId);
        
        // 创建智能线程池
        const threadPool = new AdaptiveThreadPool();
        
        // 上传任务处理
        while (!this.partTracker.isComplete()) {
            if (this.pauseSignal) {
                await this.saveProgress();
                throw new UploadPausedException();
            }
            
            const parts = this.partTracker.getNextParts(threadPool.availableSlots());
            parts.forEach(part => {
                threadPool.submit(async () => {
                    try {
                        const etag = await this.uploadPart(part);
                        this.partTracker.completePart(part.number, etag);
                        this.autoSaveProgress();
                    } catch (err) {
                        this.partTracker.failPart(part.number);
                        this.handleError(err);
                    }
                });
            });
            
            await sleep(100); // 避免CPU空转
        }
        
        // 完成上传
        await this.completeUpload();
    }
    
    pause() { this.pauseSignal = true; }
    resume() { this.pauseSignal = false; this.startUpload(); }
}

public class AdaptiveThreadPool {
    private ThreadPoolExecutor executor;
    private NetworkMonitor networkMonitor;
    
    public AdaptiveThreadPool() {
        this.networkMonitor = new NetworkMonitor();
        this.executor = new ThreadPoolExecutor(
            4, // 核心线程数
            32, // 最大线程数
            60, TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000)
        );
        
        // 启动监控线程
        new Thread(this::monitorAndAdjust).start();
    }
    
    private void monitorAndAdjust() {
        while (true) {
            // 基于网络状况调整
            double packetLoss = networkMonitor.getPacketLossRate();
            if (packetLoss > 0.1) {
                executor.setCorePoolSize(4); // 高丢包时减少并发
            } else {
                int suggested = (int)(NetworkMonitor.getAvailableBandwidth() / (5 * 1024));
                executor.setCorePoolSize(Math.min(32, Math.max(4, suggested)));
            }
            
            // 基于队列深度调整
            if (executor.getQueue().size() > 500) {
                executor.setMaximumPoolSize(Math.min(64, executor.getMaximumPoolSize() + 4));
            }
            
            Thread.sleep(5000); // 每5秒调整一次
        }
    }
}

def calculate_dynamic_part_size(file_size, network_quality):
    """
    基于文件大小和网络状况的动态分片算法
    :param file_size: 文件大小(bytes)
    :param network_quality: 网络质量评分(0-1)
    :return: 最优分片大小(bytes)
    """
    # 基础分片大小
    base = 5 * 1024 * 1024  # 5MB
    
    # 根据文件大小调整
    if file_size > 10 * 1024 * 1024 * 1024:  # >10GB
        base = 20 * 1024 * 1024
    elif file_size > 1 * 1024 * 1024 * 1024:  # >1GB
        base = 10 * 1024 * 1024
    
    # 根据网络质量调整
    if network_quality < 0.3:  # 差网络
        return max(1 * 1024 * 1024, base / 2)
    elif network_quality > 0.8:  # 优质网络
        return min(100 * 1024 * 1024, base * 2)
    
    return base

public class SmartRetryPolicy {
    private static final int MAX_RETRIES = 5;
    private static final long BASE_DELAY = 1000; // 1s
    
    public void executeWithRetry(Runnable task) {
        int retryCount = 0;
        while (retryCount <= MAX_RETRIES) {
            try {
                task.run();
                return;
            } catch (NetworkException e) {
                retryCount++;
                long delay = calculateBackoff(retryCount);
                Thread.sleep(delay);
            } catch (NonRetriableException e) {
                throw e;
            }
        }
        throw new MaxRetriesExceededException();
    }
    
    private long calculateBackoff(int retryCount) {
        // 指数退避+随机抖动
        long expDelay = (long) Math.pow(2, retryCount) * BASE_DELAY;
        long jitter = (long) (Math.random() * 1000);
        return expDelay + jitter;
    }
}

type MemoryPool struct {
    pool chan []byte
}

func NewMemoryPool(blockSize int, maxBlocks int) *MemoryPool {
    return &MemoryPool{
        pool: make(chan []byte, maxBlocks),
    }
}

func (p *MemoryPool) Get() []byte {
    select {
    case buf := <-p.pool:
        return buf
    default:
        return make([]byte, blockSize)
    }
}

func (p *MemoryPool) Put(buf []byte) {
    select {
    case p.pool <- buf:
    default: // 池已满，丢弃缓冲区
    }
}

class AITuner:
    def optimize_parameters(self, file_size, network_stats, hw_spec):
        # 使用强化学习模型预测最优参数
        model = load_model("upload_optimizer.h5")
        return model.predict([file_size, 
                             network_stats.latency, 
                             network_stats.bandwidth,
                             hw_spec.cpu_cores,
                             hw_spec.memory])

+---------------------+---------------------+
| 传统TCP上传         | QUIC加速上传        |
+---------------------+---------------------+
| 3次握手建立连接     | 0-RTT快速启动       |
| 队头阻塞问题        | 多路复用无阻塞      |
| 拥塞控制反应慢      | 改进的拥塞算法      |
| 移动网络切换中断    | 连接迁移支持        |
+---------------------+---------------------+

#!/bin/bash
# oss_benchmark.sh
FILE_SIZES=("10m" "100m" "1g" "10g")
THREADS=(4 8 16 32)
METHODS=("single" "multipart" "resumable")

for size in "${FILE_SIZES[@]}"; do
  for thread in "${THREADS[@]}"; do
    for method in "${METHODS[@]}"; do
      echo "Testing ${size} file with ${thread} threads (${method})"
      ./upload_tool --size $size --threads $thread --method $method --output report_${size}_${thread}_${method}.json
    done
  done
done

# 生成可视化报告
python analyze_results.py

def collect_metrics():
    return {
        "timestamp": time.time(),
        "network": {
            "bandwidth": get_available_bandwidth(),
            "latency": measure_latency("oss-endpoint"),
            "packet_loss": get_packet_loss_rate()
        },
        "system": {
            "cpu_usage": psutil.cpu_percent(),
            "memory_usage": psutil.virtual_memory().percent,
            "io_wait": psutil.cpu_times().iowait
        },
        "upload": {
            "progress": current_progress,
            "current_speed": calculate_instant_speed(),
            "active_threads": threading.active_count()
        }
    }