---
id: 8wel7k12347894gsdjkfnru23kl78989
author: Yunjin Lee
title: Go HTTP 文件上传响应延迟优化:基于通道的异步工作池队列实现
description: 该结构通过缓冲通道和 goroutine 工作池，将 net/http 文件上传处理程序中的临时文件接收与 Incus 容器传输操作分离，从而降低端点的响应延迟。
language: zh
date: 2025-07-14T11:51:27.614782351Z
path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238
category: concurrency
---

# 概述

初学 Go 时，常常通过实现后端服务器来边学边用。此时，不妨以实现一个通过 RestAPI 等方式接收文件流并上传至服务器的示例为例。
Go 语言的 net/http 服务器默认会同时处理多个请求，因此并发上传本身并不存在问题。
然而，如果在接收流之后的所有操作都以同步方式进行，会导致端点响应延迟。
接下来，我们将探讨用于防止此类情况的机制。

# 原因

接收流本身通常需要较长时间，对于较大的文件，单个请求可能会处理数分钟之久。
在这种情况下，尽可能迅速地处理接收之后的操作至关重要。
此示例场景是接收流之后将其保存为临时文件，并推送至容器。
此时，如果使用 worker pool 来处理将临时文件推送至容器的部分，则可以缩短响应延迟。

```go
package file_upload

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"time"
)

const uploadTempDir = "/tmp/incus_uploads" // Host temporary directory

// UploadTask holds data for asynchronous file push.
type UploadTask struct {
	HostTempFilePath         string
	ContainerName            string
    HostFilename             string
	ContainerDestinationPath string
}

// UploadHandler processes file uploads. Saves to temp file, then queues for Incus push.
func UploadHandler(wr http.ResponseWriter, req *http.Request) {
	if req.Method != http.MethodPost {
		http.Error(wr, "POST method required.", http.StatusMethodNotAllowed)
		return
	}
	originalFilePath := req.Header.Get("X-File-Path")
    originalFilename := filepath.Base(req.Header.Get("X-Host-Path"))
	containerName := req.Header.Get("X-Container-Name")
	if originalFilePath == "" || containerName == "" {
		http.Error(wr, "Missing X-File-Path or X-Container-Name header.", http.StatusBadRequest)
		return
	}

	cleanContainerDestPath := filepath.Clean(originalFilePath)
	if !filepath.IsAbs(cleanContainerDestPath) {
		http.Error(wr, "File path must be absolute.", http.StatusBadRequest)
		return
	}

	// Create unique temporary file path on host
	tempFileName := fmt.Sprintf("%d-%s", time.Now().UnixNano(), filepath.Base(originalFilePath))
	hostTempFilePath := filepath.Join(uploadTempDir, tempFileName)

	if err := os.MkdirAll(uploadTempDir, 0755); err != nil {
		log.Printf("ERROR: Failed to create temp upload directory: %v", err)
		http.Error(wr, "Server error.", http.StatusInternalServerError)
		return
	}

	// Create and copy request body to temporary file (synchronous)
	outFile, err := os.Create(hostTempFilePath)
	if err != nil {
		log.Printf("ERROR: Failed to create temporary file: %v", err)
		http.Error(wr, "Server error.", http.StatusInternalServerError)
		return
	}
	defer outFile.Close()

	bytesWritten, err := io.Copy(outFile, req.Body)
	if err != nil {
		outFile.Close()
		os.Remove(hostTempFilePath)
		log.Printf("ERROR: Failed to copy request body to temp file: %v", err)
		http.Error(wr, "File transfer failed.", http.StatusInternalServerError)
		return
	}
	log.Printf("Upload Info: Received %d bytes, saved to %s.", bytesWritten, hostTempFilePath)

	// Enqueue task for asynchronous Incus push
	task := UploadTask{
		HostTempFilePath:         hostTempFilePath,
		ContainerName:            containerName,
        HostFilename :            originalFilename,
		ContainerDestinationPath: cleanContainerDestPath,
	}
	EnqueueTask(task) //THIS PART
	log.Printf("Upload Info: Task enqueued for %s to %s.", originalFilePath, containerName)

	// Send immediate 202 Accepted response
	wr.WriteHeader(http.StatusAccepted)
	fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
}
```

观察此处标记为 THIS PART 的部分，您应当已经注意到任务被插入到了队列中。

现在，让我们来看看任务队列是如何运作的。

```go
package file_upload

import (
	"log"
	"sync"
)

var taskQueue chan UploadTask
var once sync.Once

// InitWorkQueue initializes the in-memory task queue.
func InitWorkQueue() {
	once.Do(func() {
		taskQueue = make(chan UploadTask, 100)
		log.Println("Upload Info: Work queue initialized.")
	})
}

// EnqueueTask adds an UploadTask to the queue.
func EnqueueTask(task UploadTask) {
	if taskQueue == nil {
		log.Fatal("ERROR: Task queue not initialized.")
	}
	taskQueue <- task
	log.Printf("Upload Info: Queue: Task enqueued. Size: %d", len(taskQueue))
}

// DequeueTask retrieves an UploadTask from the queue, blocking if empty.
func DequeueTask() UploadTask {
	if taskQueue == nil {
		log.Fatal("ERROR: Task queue not initialized.")
	}
	task := <-taskQueue
	log.Printf("Upload Info: Queue: Task dequeued. Size: %d", len(taskQueue))
	return task
}

// GetQueueLength returns current queue size.
func GetQueueLength() int {
	if taskQueue == nil {
		return 0
	}
	return len(taskQueue)
}
```

作为示例提供的任务队列实现较为简单。该任务队列具有从通道中取出队列内任务的简单结构。

以下是用於在文件上传之后推送到容器的 worker 方法。
为了实现良好的响应性和简便的实现，当前方法采用了无限循环，但您也可以根据具体用途添加相应的算法。

```go
func StartFilePushWorker() {
	for {
		task := DequeueTask()
		log.Printf("Upload Info: Worker processing task for %s from %s.", task.ContainerName, task.HostFilename)

		// Defer cleanup of the temporary file
		defer func(filePath string) {
			if err := os.Remove(filePath); err != nil {
				log.Printf("ERROR: Worker: Failed to remove temp file '%s': %v", filePath, err)
			} else {
				log.Printf("Upload Info: Worker: Cleaned up temp file: %s", filePath)
			}
		}(task.HostTempFilePath)

		// Process task with retries for transient Incus errors
		for i := 0; i <= MaxRetries; i++ {
			err := processUploadTask(task) //separate upload task
			if err == nil {
				log.Printf("SUCCESS: Worker: Task completed for %s.", task.ContainerName)
				break
			}

			isTransient := true
			if err.Error() == "incus: container not found" { // Example permanent error
				isTransient = false
			}

			if isTransient && i < MaxRetries {
				log.Printf("WARNING: Worker: Task failed for %s (attempt %d/%d): %v. Retrying.",
					task.ContainerName, i+1, MaxRetries, err)
				time.Sleep(RetryDelay)
			} else {
				log.Printf("ERROR: Worker: Task permanently failed for %s after %d attempts: %v.",
					task.ContainerName, i+1, err)
				break
			}
		}
	}
}
```

首先，该函数持续循环以尝试从队列中获取任务。接着，在重试次数范围内，尝试执行从临时文件到容器的上传步骤，而非从流到临时文件的步骤。

# 优势

这种处理方式的优势在于，**只要流上传本身正常**，便能缩短后续处理任务的延迟时间，并可防止因同时发生的大量容器操作而导致资源耗竭。正如当前代码所示，可同时进行的容器操作数量受限于通道的数量。至此，我们已经了解了一个能够切实应用 Go 并发处理的示例。若想查看更多示例，请访问以下链接。
[包含示例的模块](https://github.com/gg582/linux_virt_unit)
[利用示例的项目](https://github.com/gg582/incuspeed)
由于项目本身包含许多辅助组件，关于 worker 本身的学习，只需简要浏览 main.go 中是如何调用 worker 初始化函数的即可。该模块中还包含其他形态的 worker，请一并参考。