-
Notifications
You must be signed in to change notification settings - Fork 42
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- 增加多线程分片下载文件 - 增加下载文件但是进程退出后，重新下载恢复之前的进度
- Loading branch information
Showing
7 changed files
with
280 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,101 @@ | ||
| package partial | ||
|
|
||
| import ( | ||
| "sync" | ||
| ) | ||
|
|
||
| type Chunk struct { | ||
|
|
||
| // 切片内容的在源文件的开始地址 | ||
| start int64 | ||
|
|
||
| // 切片内容在源文件的结束地址 | ||
| end int64 | ||
|
|
||
| // 切片任务的下载错误 | ||
| err error | ||
|
|
||
| // 切片的顺序 | ||
| index int | ||
|
|
||
| // 下载完的切片的具体内容 | ||
| buffer []byte | ||
| } | ||
|
|
||
| func NewChunk(index int, start, end int64) *Chunk { | ||
| chunk := &Chunk{ | ||
| start: start, | ||
| end: end, | ||
| index: index, | ||
| } | ||
| return chunk | ||
| } | ||
|
|
||
| func (p *Chunk) SetData(bytes []byte) { | ||
| p.buffer = bytes | ||
| } | ||
|
|
||
| func (p *Chunk) SetError(err error) { | ||
| p.err = err | ||
| } | ||
|
|
||
| func (p *Chunk) Error() error { | ||
| return p.err | ||
| } | ||
|
|
||
| func (p *Chunk) Data() []byte { | ||
| return p.buffer | ||
| } | ||
|
|
||
| // 切片乱序写入后，将切片顺序读取 | ||
| type ChunksSorter struct { | ||
| // 已经读取的切片数量 | ||
| readCount int | ||
|
|
||
| // 切片的所有总数 | ||
| chunkCount int | ||
|
|
||
| // 线程数，用于阻塞写入 | ||
| works int | ||
|
|
||
| // 存储切片的缓存区 | ||
| chunks []chan *Chunk | ||
|
|
||
| lock sync.Mutex | ||
| } | ||
|
|
||
| func newChunksSorter(chunkCount int, works int) *ChunksSorter { | ||
| chunks := make([]chan *Chunk, works) | ||
| for i := 0; i < len(chunks); i++ { | ||
| chunks[i] = make(chan *Chunk) | ||
| } | ||
|
|
||
| return &ChunksSorter{ | ||
| chunkCount: chunkCount, | ||
| works: works, | ||
| chunks: chunks, | ||
| } | ||
| } | ||
|
|
||
| // 判断分片是否读取完毕 | ||
| func (p *ChunksSorter) HasReadAll() bool { | ||
| defer p.lock.Unlock() | ||
| p.lock.Lock() | ||
| return p.chunkCount == p.readCount | ||
| } | ||
|
|
||
| // 将数据写入到缓存区，如果该缓存已满，则会被阻塞 | ||
| func (p *ChunksSorter) Write(chunk *Chunk) { | ||
| p.chunks[chunk.index%p.works] <- chunk | ||
| } | ||
|
|
||
| // 顺序读取切片，如果下一个切片没有下载完，则会被阻塞 | ||
| func (p *ChunksSorter) Read() *Chunk { | ||
| defer p.lock.Unlock() | ||
| p.lock.Lock() | ||
|
|
||
| chunk := <-p.chunks[p.readCount%p.works] | ||
| p.readCount++ | ||
|
|
||
| return chunk | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,113 @@ | ||
| package partial | ||
|
|
||
| import ( | ||
| "errors" | ||
| "io" | ||
| "os" | ||
| ) | ||
|
|
||
| const ChunkSize = 1024 * 1024 * 10 | ||
|
|
||
| type ChunkDownFunc func(start, end int64) ([]byte, error) | ||
|
|
||
| type MultiPartialDownloader struct { | ||
|
|
||
| // 文件路径 | ||
| filePath string | ||
|
|
||
| // 最终文件大小 | ||
| finalSize int64 | ||
|
|
||
| // 本地文件大小 | ||
| localSize int64 | ||
|
|
||
| writer io.Writer | ||
| works int | ||
| downFunc ChunkDownFunc | ||
| } | ||
|
|
||
| func NewMultiPartialDownloader(filePath string, finalSize int64, writer io.Writer, works int, fn ChunkDownFunc) *MultiPartialDownloader { | ||
| return &MultiPartialDownloader{ | ||
| filePath: filePath, | ||
| finalSize: finalSize, | ||
| works: works, | ||
| writer: writer, | ||
| downFunc: fn, | ||
| } | ||
| } | ||
|
|
||
| func (p *MultiPartialDownloader) Download() error { | ||
| fileinfo, err := os.Stat(p.filePath) | ||
| if err == nil { | ||
| p.localSize = fileinfo.Size() | ||
| } | ||
|
|
||
| // 计算需要下载的块数 | ||
| needDownSize := p.finalSize - p.localSize | ||
| chunkCount := needDownSize / ChunkSize | ||
| if needDownSize%ChunkSize != 0 { | ||
| chunkCount++ | ||
| } | ||
|
|
||
| // 发布任务队列 | ||
| var queue = make(chan *Chunk, p.works) | ||
| go func() { | ||
| defer close(queue) | ||
| for i := 0; i < int(chunkCount); i++ { | ||
| start := p.localSize + int64(i)*ChunkSize | ||
| end := p.localSize + int64(i+1)*ChunkSize | ||
| if end > p.finalSize { | ||
| end = p.finalSize | ||
| } | ||
| queue <- NewChunk(i, start, end) | ||
| } | ||
| }() | ||
|
|
||
| chunksSorter := newChunksSorter( | ||
| int(chunkCount), | ||
| p.works, | ||
| ) | ||
|
|
||
| // 下载切片任务 | ||
| for i := 0; i < int(p.works); i++ { | ||
| go func() { | ||
| for chunk := range queue { | ||
| var ( | ||
| err error | ||
| buffer []byte | ||
| ) | ||
|
|
||
| // 重试三次 | ||
| for t := 0; t < 3; t++ { | ||
| // ? 由于长度是从1开始，而数据是从0地址开始 | ||
| // ? 计算字节时容量会多出开头的一位，所以末尾需要减少一位 | ||
| buffer, err = p.downFunc(chunk.start, chunk.end-1) | ||
| if err == nil { | ||
| break | ||
| } | ||
| } | ||
| chunk.SetData(buffer) | ||
| chunk.SetError(err) | ||
| chunksSorter.Write(chunk) | ||
| } | ||
| }() | ||
| } | ||
|
|
||
| // 将分片顺序写入到文件 | ||
| for { | ||
| if chunksSorter.HasReadAll() { | ||
| break | ||
| } | ||
|
|
||
| chunk := chunksSorter.Read() | ||
| if chunk.Error() != nil { | ||
| return chunk.Error() | ||
| } | ||
|
|
||
| if len(chunk.Data()) == 0 { | ||
| return errors.New("chunk buffer download but size is 0") | ||
| } | ||
| p.writer.Write(chunk.Data()) | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.