Go语言实现多协程文件下载器的过程详解_Golang

前言

你好，我是醉墨居士，最近在开发文件传输相关的项目，然后顺手写了一个多协程文件下载器，代码非常精简，核心代码只有100行左右，适合分享给大家学习使用

流程图

主函数

func main() {
	fileurl := flag.string("u", "", "downloade url of the file")
	flag.parse()
	if *fileurl == "" {
		log.println("please input a download url")
		flag.usage()
		return
	}
	filedir, err := os.getwd()
	if err != nil {
		log.println(err)
		return
	}
	// 下载文件保存路径
	filepath := filepath.join(filedir, filepath.base(*fileurl))
	err = downloadfile(*fileurl, filepath)
	if err != nil {
		log.println(err)
		return
	}
	log.println("download file success:", filepath)
}

下载文件

// 下载文件
func downloadfile(fileurl string, filepath string) error {
	log.println("downloading file:", fileurl, "to", filepath)
	taskch := make(chan [2]int64, runtime.numcpu())
	wg := new(sync.waitgroup)
	// 创建执行下载任务的 worker
	err := initworker(fileurl, filepath, taskch, wg)
	if err != nil {
		return fmt.errorf("init worker failed: %v", err)
	}
	// 分发下载任务
	err = dispatchtask(fileurl, taskch)
	if err != nil {
		return fmt.errorf("dispacth task failed: %v", err)
	}
	// 等待所有下载任务完成
	wg.wait()
	return nil
}

初始化分片下载worker

// 初始化 下载 worker
func initworker(url string, filepath string, taskch chan [2]int64, wg *sync.waitgroup) error {
	for i := 0; i < runtime.numcpu(); i++ {
		// 打开文件句柄
		file, err := os.openfile(filepath, os.o_create|os.o_rdwr, 0644)
		if err != nil {
			return err
		}
		wg.add(1)
		go func(file *os.file, taskch chan [2]int64) {
			defer wg.done()
			defer file.close()
			// 循环从 taskch 中获取下载任务并下载
			for part := range taskch {
				log.printf("downloading part, start offset: %d, end offset: %d", part[0], part[1])
				// 重试下载，最大重试次数为 10 次，每次下载失败后等待 1 秒
				err := retrywithwaittime(10, func() error {
					return downloadpart(url, file, part[0], part[1])
				}, time.second)
				if err != nil {
					log.printf("download part %d failed: %v", part, err)
				}
			}
		}(file, taskch)
	}
	return nil
}

分发下载任务

// 分发下载任务
func dispatchtask(url string, taskch chan [2]int64) error {
	defer close(taskch)
	filesize, err := getfilesize(url)
	if err != nil {
		return err
	}
	// 分片大小 1mb
	const chunksize = 1024 * 1024
	parts := filesize / chunksize
	log.println("file size:", filesize, "parts:", parts, "chunk size:", chunksize)
	for i := int64(0); i < parts; i++ {
		// 计算分片的起始和结束位置
		startoffset := i * chunksize
		endoffset := startoffset + chunksize - 1
		// 发送下载任务
		taskch <- [2]int64{startoffset, endoffset}
	}
	// 发送最后一个分片的下载任务
	if filesize % chunksize != 0 {
		taskch <- [2]int64{parts * chunksize, filesize - 1}
	}
	return nil
}

获取下载文件的大小

// 获取文件大小
func getfilesize(url string) (int64, error) {
	resp, err := http.head(url)
	if err != nil {
		return 0, err
	}
	defer resp.body.close()
	return resp.contentlength, nil
}

下载文件分片

// 下载文件分片
func downloadpart(url string, file *os.file, startpos, endpos int64) error {
	req, err := http.newrequest("get", url, nil)
	if err != nil {
		return err
	}
	// 设置文件分片区间的请求头
	req.header.set("range", fmt.sprintf("bytes=%d-%d", startpos, endpos))
	resp, err := http.defaulttransport.roundtrip(req)
	if err != nil {
		return err
	}
	defer resp.body.close()
	// 如果服务器返回的状态码不是 206 partial content，则说明下载失败
	if resp.statuscode != http.statuspartialcontent {
		data, err := io.readall(resp.body)
		if err != nil {
			return err
		}
		log.println("unexpected data:", string(data))
		return fmt.errorf("unexpected status code: %d", resp.statuscode)
	}
	// 文件指针移动到分片的起始位置
	_, err = file.seek(startpos, 0)
	if err != nil {
		return err
	}
	// 写入分片数据到文件
	_, err = io.copy(file, resp.body)
	if err != nil {
		return err
	}
	return nil
}

错误重试

// 重试函数
func retrywithwaittime(retrycount int, fn func() error, waittime time.duration) error {
	var err error
	for i := 0; i < retrycount; i++ {
		e := fn()
		if e != nil {
			errors.join(err, e)
			time.sleep(waittime)
			continue
		}
		return nil
	}
	return err
}

项目演示