Golang重构wget脚本构建通用并发下载工具的实践指南

type Downloader interface {
    Download(url, filepath string) error
    SupportsProtocol(protocol string) bool
}

type HTTPDownloader struct {
    Client *http.Client
}

type FTPDownloader struct {
    Timeout time.Duration
}

type DownloadTask struct {
    URL      string
    FilePath string
    FileSize int64
    Parts    []Part
}

type Part struct {
    Start  int64
    End    int64
    Index  int
    Data   []byte
}

func CreateDownloader(url string) (Downloader, error) {
    u, err := url.Parse(url)
    if err != nil {
        return nil, err
    }
    
    switch u.Scheme {
    case "http", "https":
        return &HTTPDownloader{
            Client: &http.Client{Timeout: 30 * time.Second},
        }, nil
    case "ftp":
        return &FTPDownloader{
            Timeout: 30 * time.Second,
        }, nil
    default:
        return nil, fmt.Errorf("不支持的协议: %s", u.Scheme)
    }
}

func (m *Manager) Download(task *DownloadTask) error {
    // 1. 获取文件信息
    info, err := m.getFileInfo(task.URL)
    if err != nil {
        return err
    }
    task.FileSize = info.Size
    
    // 2. 检查服务器是否支持分块下载
    if !info.SupportsPartial {
        return m.simpleDownload(task)
    }
    
    // 3. 分块并发下载
    return m.concurrentDownload(task)
}

func (m *Manager) concurrentDownload(task *DownloadTask) error {
    // 创建有限数量的worker
    semaphore := make(chan struct{}, m.MaxConcurrent)
    var wg sync.WaitGroup
    errCh := make(chan error, len(task.Parts))
    
    for i, part := range task.Parts {
        wg.Add(1)
        semaphore <- struct{}{} // 获取信号量
        
        go func(idx int, p Part) {
            defer wg.Done()
            defer func() { <-semaphore }() // 释放信号量
            
            if err := m.downloadPart(task, p); err != nil {
                errCh <- fmt.Errorf("分块%d下载失败: %v", idx, err)
            }
        }(i, part)
    }
    
    wg.Wait()
    close(errCh)
    
    // 处理错误
    for err := range errCh {
        m.logger.Error("下载错误", "error", err)
    }
    
    return nil
}

func calculateChunks(fileSize int64, maxConcurrent int) []Part {
    var parts []Part
    chunkSize := calculateOptimalChunkSize(fileSize, maxConcurrent)
    
    for i := int64(0); i < fileSize; i += chunkSize {
        start := i
        end := start + chunkSize - 1
        if end >= fileSize {
            end = fileSize - 1
        }
        
        parts = append(parts, Part{
            Start: start,
            End:   end,
            Index: len(parts),
        })
    }
    
    return parts
}

type Progress struct {
    URL     string    `json:"url"`
    FilePath string  `json:"file_path"`
    FileSize int64   `json:"file_size"`
    Downloaded int64 `json:"downloaded"`
    Parts    []PartProgress `json:"parts"`
}

func (m *Manager) ResumeDownload(progressFile string) error {
    // 从进度文件恢复下载状态
    progress, err := m.loadProgress(progressFile)
    if err != nil {
        return err
    }
    
    // 只下载未完成的分块
    for i, part := range progress.Parts {
        if !part.Completed {
            go m.downloadPart(progress.Task, part.Part)
        }
    }
    
    return nil
}

func (m *Manager) downloadWithRetry(task *DownloadTask, part Part, maxRetries int) error {
    var lastErr error
    
    for retry := 0; retry < maxRetries; retry++ {
        if err := m.downloadPart(task, part); err != nil {
            lastErr = err
            m.logger.Warn("下载失败，准备重试", 
                "part", part.Index, "retry", retry+1, "error", err)
            
            // 指数退避
            time.Sleep(time.Duration(math.Pow(2, float64(retry))) * time.Second)
            continue
        }
        
        return nil
    }
    
    return fmt.Errorf("分块%d下载失败，最大重试次数已达: %v", part.Index, lastErr)
}

package main

import (
    "encoding/json"
    "flag"
    "fmt"
    "log"
    "os"
    "path/filepath"
    "time"
)

type Config struct {
    Downloads []DownloadTask `json:"downloads"`
    MaxConcurrent int       `json:"max_concurrent"`
    OutputDir    string     `json:"output_dir"`
    RetryCount   int        `json:"retry_count"`
}

func main() {
    configFile := flag.String("c", "downloads.json", "配置文件路径")
    outputReport := flag.String("r", "download_report.json", "报告文件路径")
    flag.Parse()
    
    // 读取配置文件
    config, err := loadConfig(*configFile)
    if err != nil {
        log.Fatal("读取配置文件失败:", err)
    }
    
    // 创建下载管理器
    manager := NewManager(ManagerConfig{
        MaxConcurrent: config.MaxConcurrent,
        OutputDir:     config.OutputDir,
        RetryCount:    config.RetryCount,
    })
    
    // 执行下载任务
    results := manager.DownloadAll(config.Downloads)
    
    // 生成下载报告
    report := GenerateReport(results)
    if err := SaveReport(*outputReport, report); err != nil {
        log.Printf("生成报告失败: %v", err)
    }
    
    // 输出摘要信息
    fmt.Printf("下载完成! 成功: %d, 失败: %d, 报告: %s\n",
        report.SuccessCount, report.FailureCount, *outputReport)
}

{
    "max_concurrent": 5,
    "output_dir": "./downloads",
    "retry_count": 3,
    "downloads": [
        {
            "url": "ftp://deploy:deploy.1@222.128.9.137:6041/software/ai_monitor/latest/dog.v2.0.5.tar.gz",
            "file_path": "ai_monitor/dog.v2.0.5.tar.gz"
        },
        {
            "url": "ftp://deploy:deploy.1@222.128.9.137:6041/software/ai_op/latest/opwebmd_v1.3.64.tar.gz", 
            "file_path": "ai_op/opwebmd_v1.3.64.tar.gz"
        }
    ]
}