Method processUploadTask

api/service/batch_upload_manager.go:308–359

processUploadTask 处理上传任务

(task *UploadTask)

// processUploadTask 处理上传任务
func (m *BatchUploadManager) processUploadTask(task *UploadTask) {
	defer func() { <-m.semaphore }()

	batch, err := m.GetBatch(task.BatchID)
	if err != nil {
		return
	}

	// 获取 fileUpload 的安全副本并使用线程安全的更新方法，避免在持有 batch.mu 时重复加锁导致死锁
	fileUploadPtr, exists := batch.GetFileUpload(task.FileID)
	if !exists || fileUploadPtr == nil {
		return
	}
	if m.IsBatchCancelled(task.BatchID) {
		return
	}
	// 使用副本修改并通过 updateFileProgress 持久化，避免在同一 goroutine 中重复持有 mutex 导致死锁
	fileUpload := *fileUploadPtr
	// 更新文件状态为上传中
	fileUpload.Status = "uploading"
	fileUpload.StartTime = time.Now()
	// 生成10-30之间的随机进度值
	if progress, err := rand.Int(rand.Reader, big.NewInt(21)); err == nil {
		fileUpload.Progress = float64(10 + progress.Int64())
	} else {
		fileUpload.Progress = 20 // 出错时使用默认值
	}
	// 使用线程安全的更新接口（内部对 batch.mu 加锁）
	m.updateFileProgress(task.BatchID, task.FileID, &fileUpload)

	// 执行上传
	err = m.executeUpload(task)
	if err != nil {
		if errors.Is(err, errBatchUploadCancelled) {
			return
		}
		fileUpload.Status = "failed"
		fileUpload.Error = err.Error()
		fileUpload.Progress = 0
		batch.FailedFiles++
		m.updateFileProgress(task.BatchID, task.FileID, &fileUpload)
	} else {
		if m.IsBatchCancelled(task.BatchID) {
			return
		}
		// executeUpload已经处理了所有状态更新，这里只需要更新批次计数
		batch.UploadedFiles++
	}

	fileUpload.UpdateTime = time.Now()
	m.updateBatchStatus(batch)
}

// executeUpload 执行文件上传
func (m *BatchUploadManager) executeUpload(task *UploadTask) error {