Add multi-part concurrent download with single-stream fallback

README.md

@@ -13,6 +13,7 @@ pinned: false
 
 - 在沙箱根目录内浏览、上传、删除、移动、重命名文件
 - 发起服务端 URL 下载任务
+- URL 下载支持 Range 分片并发加速（服务器不支持时自动回退单线程）
 - 发起压缩包解压任务（支持 `.zip`、`.tar.gz`、`.tgz`）
 - 使用 SQLite 持久化任务状态并展示进度
 

backend/internal/services/tasks.go

@@ -14,6 +14,7 @@ import (
 	"path/filepath"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"fastfileviewer/backend/internal/models"
@@ -38,6 +39,12 @@ type TaskService struct {
 	subscribers map[uint]map[chan taskEvent]struct{}
 }
 
+const (
+	minMultiPartSize = 16 * 1024 * 1024
+	maxDownloadParts = 8
+	partRetryLimit   = 3
+)
+
 func NewTaskService(database *gorm.DB, fs *FSService, workers int) *TaskService {
 	if workers < 1 {
 		workers = 1
@@ -170,7 +177,52 @@ func (s *TaskService) handleDownload(task *models.Task) error {
 	}
 
 	s.appendLog(task, "starting download: "+task.Source)
-	resp, err := http.Get(task.Source) // #nosec G107 user-provided URLs are expected in this app design
+	size, supportsRange, err := probeDownload(task.Source)
+	if err == nil && supportsRange && size >= minMultiPartSize {
+		parts := int(size / (8 * 1024 * 1024))
+		if parts < 2 {
+			parts = 2
+		}
+		if parts > maxDownloadParts {
+			parts = maxDownloadParts
+		}
+		s.appendLog(task, fmt.Sprintf("range download enabled: parts=%d, size=%d bytes", parts, size))
+		if err := s.multiPartDownload(task, task.Source, target, size, parts); err == nil {
+			s.appendLog(task, "download complete")
+			return nil
+		} else {
+			s.appendLog(task, "multi-part failed, fallback to single stream: "+err.Error())
+			_ = os.Remove(target)
+		}
+	}
+
+	if err := s.singlePartDownload(task, task.Source, target); err != nil {
+		return err
+	}
+	s.appendLog(task, "download complete")
+	return nil
+}
+
+func probeDownload(url string) (size int64, supportsRange bool, err error) {
+	req, err := http.NewRequest(http.MethodHead, url, nil)
+	if err != nil {
+		return 0, false, err
+	}
+	resp, err := http.DefaultClient.Do(req) // #nosec G107 user-provided URLs are expected in this app design
+	if err != nil {
+		return 0, false, err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode < 200 || resp.StatusCode >= 400 {
+		return 0, false, fmt.Errorf("probe failed: %s", resp.Status)
+	}
+	size = resp.ContentLength
+	supportsRange = strings.Contains(strings.ToLower(resp.Header.Get("Accept-Ranges")), "bytes")
+	return size, supportsRange, nil
+}
+
+func (s *TaskService) singlePartDownload(task *models.Task, url, target string) error {
+	resp, err := http.Get(url) // #nosec G107 user-provided URLs are expected in this app design
 	if err != nil {
 		return err
 	}
@@ -191,7 +243,6 @@ func (s *TaskService) handleDownload(task *models.Task) error {
 	reader := bufio.NewReader(resp.Body)
 	buf := make([]byte, 256*1024)
 	lastTick := time.Now()
-
 	for {
 		n, readErr := reader.Read(buf)
 		if n > 0 {
@@ -215,8 +266,120 @@ func (s *TaskService) handleDownload(task *models.Task) error {
 			return readErr
 		}
 	}
+	return nil
+}
 
-	s.appendLog(task, "download complete")
+func (s *TaskService) multiPartDownload(task *models.Task, url, target string, total int64, parts int) error {
+	file, err := os.OpenFile(target, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o644)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+	if err := file.Truncate(total); err != nil {
+		return err
+	}
+
+	var downloaded atomic.Int64
+	stopProgress := make(chan struct{})
+	go func() {
+		ticker := time.NewTicker(500 * time.Millisecond)
+		defer ticker.Stop()
+		for {
+			select {
+			case <-stopProgress:
+				return
+			case <-ticker.C:
+				progress := int((downloaded.Load() * 100) / total)
+				if progress > 99 {
+					progress = 99
+				}
+				s.setProgress(task, progress)
+			}
+		}
+	}()
+
+	var wg sync.WaitGroup
+	errCh := make(chan error, parts)
+	chunkSize := total / int64(parts)
+
+	for i := 0; i < parts; i++ {
+		start := int64(i) * chunkSize
+		end := start + chunkSize - 1
+		if i == parts-1 {
+			end = total - 1
+		}
+
+		wg.Add(1)
+		go func(partID int, rangeStart, rangeEnd int64) {
+			defer wg.Done()
+			var lastErr error
+			for attempt := 1; attempt <= partRetryLimit; attempt++ {
+				err := downloadRange(url, file, rangeStart, rangeEnd, &downloaded)
+				if err == nil {
+					return
+				}
+				lastErr = err
+				time.Sleep(time.Duration(attempt) * 400 * time.Millisecond)
+			}
+			errCh <- fmt.Errorf("part %d failed: %w", partID, lastErr)
+		}(i+1, start, end)
+	}
+
+	wg.Wait()
+	close(stopProgress)
+	close(errCh)
+	for err := range errCh {
+		if err != nil {
+			return err
+		}
+	}
+	s.setProgress(task, 99)
+	return nil
+}
+
+func downloadRange(url string, out *os.File, start, end int64, downloaded *atomic.Int64) error {
+	req, err := http.NewRequest(http.MethodGet, url, nil)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", start, end))
+
+	resp, err := http.DefaultClient.Do(req) // #nosec G107 user-provided URLs are expected in this app design
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusPartialContent {
+		return fmt.Errorf("server does not honor range request: %s", resp.Status)
+	}
+
+	reader := bufio.NewReader(resp.Body)
+	buf := make([]byte, 256*1024)
+	offset := start
+	expected := end - start + 1
+	var written int64
+
+	for {
+		n, readErr := reader.Read(buf)
+		if n > 0 {
+			if _, err := out.WriteAt(buf[:n], offset); err != nil {
+				return err
+			}
+			offset += int64(n)
+			written += int64(n)
+			downloaded.Add(int64(n))
+		}
+		if readErr == io.EOF {
+			break
+		}
+		if readErr != nil {
+			return readErr
+		}
+	}
+	if written != expected {
+		return fmt.Errorf("range incomplete: expected %d got %d", expected, written)
+	}
 	return nil
 }