在Golang中实现文件下载限速，核心是通过控制单位时间内读取并写入的数据量来达成速率限制目标，结合并发场景还需要做好协程间的速率分配与同步。下面我们从基础原理到实践代码逐步讲解实现方式。
核心实现原理
文件下载的本质是从网络流读取数据，再写入本地文件。要实现限速，只需要在读取数据的过程中加入速率控制逻辑即可，常用的方案是令牌桶算法：预先设定每秒生成的令牌数量，每次读取数据前先获取对应数据量的令牌，获取不到则等待，这样就能保证单位时间内的数据读取量不超过设定值。
在Golang中我们可以使用golang.org/x/time/rate包提供的令牌桶实现，也可以自己基于time.Ticker实现简单的速率控制逻辑。
单文件下载限速实现
首先实现基础的单文件下载限速功能，这里我们使用rate.Limiter来控制速率，设定每秒允许传输的字节数，比如限制为1024*100字节每秒也就是100KB/s。
package main
import (
"context"
"fmt"
"io"
"net/http"
"os"
"golang.org/x/time/rate"
)
// 下载文件并限速
func downloadFileWithLimit(url, savePath string, limitBytesPerSec int64) error {
// 创建HTTP请求
resp, err := http.Get(url)
if err != nil {
return fmt.Errorf("请求文件失败: %v", err)
}
defer resp.Body.Close()
// 创建本地文件
file, err := os.Create(savePath)
if err != nil {
return fmt.Errorf("创建文件失败: %v", err)
}
defer file.Close()
// 创建速率限制器，每秒生成limitBytesPerSec个令牌，最大突发量为limitBytesPerSec
limiter := rate.NewLimiter(rate.Limit(limitBytesPerSec), int(limitBytesPerSec))
// 缓冲区大小，每次读取4096字节
buf := make([]byte, 4096)
ctx := context.Background()
for {
// 从响应体读取数据
n, err := resp.Body.Read(buf)
if n > 0 {
// 等待获取对应数据量的令牌
err := limiter.WaitN(ctx, n)
if err != nil {
return fmt.Errorf("等待令牌失败: %v", err)
}
// 写入本地文件
_, err = file.Write(buf[:n])
if err != nil {
return fmt.Errorf("写入文件失败: %v", err)
}
}
// 读取结束则退出循环
if err == io.EOF {
break
}
if err != nil {
return fmt.Errorf("读取数据失败: %v", err)
}
}
return nil
}
func main() {
// 示例：下载文件，限制速率为100KB/s
url := "http://ipipp.com/testfile.zip"
savePath := "./testfile.zip"
// 100KB/s = 100 * 1024 字节每秒
limit := int64(100 * 1024)
err := downloadFileWithLimit(url, savePath, limit)
if err != nil {
fmt.Printf("下载失败: %vn", err)
} else {
fmt.Println("下载完成")
}
}
并发下载的速率控制
当需要实现多文件并发下载，并且整体下载速率不超过设定值时，需要把速率限制的逻辑放在全局层面，多个下载协程共享同一个速率限制器，或者使用带权重的分配方式给每个协程分配速率配额。
下面是实现多文件并发下载并控制整体速率的示例，整体限制为200KB/s，3个文件并发下载：
package main
import (
"context"
"fmt"
"io"
"net/http"
"os"
"sync"
"golang.org/x/time/rate"
)
// 并发下载任务
type DownloadTask struct {
URL string
SavePath string
}
// 执行单个下载任务，使用共享的速率限制器
func doDownload(task DownloadTask, limiter *rate.Limiter, wg *sync.WaitGroup) {
defer wg.Done()
resp, err := http.Get(task.URL)
if err != nil {
fmt.Printf("任务%s请求失败: %vn", task.SavePath, err)
return
}
defer resp.Body.Close()
file, err := os.Create(task.SavePath)
if err != nil {
fmt.Printf("任务%s创建文件失败: %vn", task.SavePath, err)
return
}
defer file.Close()
buf := make([]byte, 4096)
ctx := context.Background()
for {
n, err := resp.Body.Read(buf)
if n > 0 {
// 等待获取令牌，所有协程共享同一个限制器，整体速率会被控制
err := limiter.WaitN(ctx, n)
if err != nil {
fmt.Printf("任务%s等待令牌失败: %vn", task.SavePath, err)
return
}
_, err = file.Write(buf[:n])
if err != nil {
fmt.Printf("任务%s写入文件失败: %vn", task.SavePath, err)
return
}
}
if err == io.EOF {
break
}
if err != nil {
fmt.Printf("任务%s读取数据失败: %vn", task.SavePath, err)
return
}
}
fmt.Printf("任务%s下载完成n", task.SavePath)
}
func main() {
// 定义下载任务列表
tasks := []DownloadTask{
{URL: "http://ipipp.com/file1.zip", SavePath: "./file1.zip"},
{URL: "http://ipipp.com/file2.zip", SavePath: "./file2.zip"},
{URL: "http://ipipp.com/file3.zip", SavePath: "./file3.zip"},
}
// 整体速率限制为200KB/s
totalLimit := int64(200 * 1024)
limiter := rate.NewLimiter(rate.Limit(totalLimit), int(totalLimit))
var wg sync.WaitGroup
for _, task := range tasks {
wg.Add(1)
go doDownload(task, limiter, &wg)
}
wg.Wait()
fmt.Println("所有下载任务完成")
}
注意事项
- 速率限制的数值需要根据实际业务场景设定，避免设置过低导致下载效率过低，或者过高失去限速意义。
- 如果下载的是小文件，令牌桶的突发量可以适当调大，避免小文件下载时等待令牌带来的额外耗时。
- 并发下载时如果需要对单个文件也做速率限制，可以在每个协程内部再加一个独立的速率限制器，实现双层限速。
- 处理网络请求时需要做好超时控制，避免下载卡住影响整体流程，可以在
http.Client中设置Timeout字段。
自定义简单限速器实现
如果不想依赖第三方包，也可以基于time.Ticker实现一个简单的固定速率限速器，适合简单的限速场景：
package main
import (
"context"
"fmt"
"io"
"net/http"
"os"
"time"
)
// 简单限速器，每秒最多传输limit字节
type SimpleLimiter struct {
limit int64 // 每秒限制字节数
bufSize int64 // 缓冲区大小
lastTime time.Time // 上次写入时间
}
func NewSimpleLimiter(limit int64) *SimpleLimiter {
return &SimpleLimiter{
limit: limit,
bufSize: 4096,
lastTime: time.Now(),
}
}
// 等待直到可以写入n字节
func (l *SimpleLimiter) WaitN(ctx context.Context, n int) error {
// 计算需要等待的时间：n字节需要 n/limit 秒
needTime := time.Duration(float64(n)/float64(l.limit)) * time.Second
elapsed := time.Since(l.lastTime)
if elapsed < needTime {
waitTime := needTime - elapsed
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(waitTime):
l.lastTime = time.Now()
return nil
}
}
l.lastTime = time.Now()
return nil
}
func downloadWithSimpleLimit(url, savePath string, limit int64) error {
resp, err := http.Get(url)
if err != nil {
return err
}
defer resp.Body.Close()
file, err := os.Create(savePath)
if err != nil {
return err
}
defer file.Close()
limiter := NewSimpleLimiter(limit)
buf := make([]byte, 4096)
ctx := context.Background()
for {
n, err := resp.Body.Read(buf)
if n > 0 {
err := limiter.WaitN(ctx, n)
if err != nil {
return err
}
_, err = file.Write(buf[:n])
if err != nil {
return err
}
}
if err == io.EOF {
break
}
if err != nil {
return err
}
}
return nil
}
func main() {
err := downloadWithSimpleLimit("http://ipipp.com/test.txt", "./test.txt", 1024*50) // 50KB/s
if err != nil {
fmt.Printf("下载失败: %vn", err)
} else {
fmt.Println("下载完成")
}
}