跳到主要内容

实现并发下载器

问题​

如何用 Go 实现一个支持多线程分片下载的文件下载器？

答案​

核心思路​

  1. HEAD 请求获取文件大小和是否支持 Range
  2. 将文件分成 N 个分片，并发下载
  3. 合并所有分片

完整实现​

type Downloader struct {
    URL         string
    Concurrency int
    OutputPath  string
}

func (d *Downloader) Download() error {
    // 1. 获取文件信息
    resp, err := http.Head(d.URL)
    if err != nil {
        return err
    }
    fileSize := resp.ContentLength
    supportsRange := resp.Header.Get("Accept-Ranges") == "bytes"

    // 不支持 Range，单线程下载
    if !supportsRange || fileSize <= 0 {
        return d.downloadSingle()
    }

    // 2. 计算分片
    chunkSize := fileSize / int64(d.Concurrency)
    chunks := make([]chunk, d.Concurrency)
    for i := 0; i < d.Concurrency; i++ {
        start := int64(i) * chunkSize
        end := start + chunkSize - 1
        if i == d.Concurrency-1 {
            end = fileSize - 1 // 最后一片取到末尾
        }
        chunks[i] = chunk{Index: i, Start: start, End: end}
    }

    // 3. 并发下载，errgroup 控制并发
    g, ctx := errgroup.WithContext(context.Background())
    tempFiles := make([]string, d.Concurrency)

    for _, ch := range chunks {
        ch := ch
        g.Go(func() error {
            return d.downloadChunk(ctx, ch, &tempFiles[ch.Index])
        })
    }

    if err := g.Wait(); err != nil {
        return fmt.Errorf("下载失败: %w", err)
    }

    // 4. 合并分片
    return d.mergeChunks(tempFiles)
}

type chunk struct {
    Index int
    Start int64
    End   int64
}

func (d *Downloader) downloadChunk(ctx context.Context, ch chunk, tempPath *string) error {
    req, _ := http.NewRequestWithContext(ctx, "GET", d.URL, nil)
    // Range 头：指定下载的字节范围
    req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", ch.Start, ch.End))

    resp, err := http.DefaultClient.Do(req)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    // 写入临时文件
    *tempPath = fmt.Sprintf("%s.part%d", d.OutputPath, ch.Index)
    f, err := os.Create(*tempPath)
    if err != nil {
        return err
    }
    defer f.Close()

    _, err = io.Copy(f, resp.Body)
    return err
}

func (d *Downloader) mergeChunks(tempFiles []string) error {
    outFile, err := os.Create(d.OutputPath)
    if err != nil {
        return err
    }
    defer outFile.Close()

    for _, path := range tempFiles {
        f, err := os.Open(path)
        if err != nil {
            return err
        }
        io.Copy(outFile, f)
        f.Close()
        os.Remove(path) // 删除临时文件
    }
    return nil
}

func (d *Downloader) downloadSingle() error {
    resp, err := http.Get(d.URL)
    if err != nil {
        return err
    }
    defer resp.Body.Close()

    f, _ := os.Create(d.OutputPath)
    defer f.Close()
    _, err = io.Copy(f, resp.Body)
    return err
}

使用示例​

func main() {
    dl := &Downloader{
        URL:         "https://example.com/large-file.zip",
        Concurrency: 8,
        OutputPath:  "large-file.zip",
    }
    if err := dl.Download(); err != nil {
        log.Fatal(err)
    }
    fmt.Println("下载完成")
}

进度显示​

// 用 io.TeeReader 或包装 Writer 统计已下载字节
type ProgressWriter struct {
    Total     int64
    Current   int64
    OnProgress func(current, total int64)
}

func (pw *ProgressWriter) Write(p []byte) (int, error) {
    n := len(p)
    atomic.AddInt64(&pw.Current, int64(n))
    if pw.OnProgress != nil {
        pw.OnProgress(atomic.LoadInt64(&pw.Current), pw.Total)
    }
    return n, nil
}

常见面试问题​

Q1: 为什么分片下载比单线程快？​

答案：

  • 利用多连接并行传输，充分利用带宽
  • 单连接受 TCP 拥塞窗口限制
  • CDN 可能对单连接限速

Q2: 某个分片下载失败怎么办？​

答案：

  • errgroup 会取消其他 goroutine 的 Context
  • 可以为单个分片加重试（指数退避）
  • 支持断点续传：记录每个分片进度，失败后从上次位置继续

相关链接​