---
id: 8wel7k12347894gsdjkfnru23kl78989
author: Yunjin Lee
title: 'Намаляване на забавянето при отговорите за качване на файлове през Go HTTP: реализация на асинхронна работна опашка, базирана на канали'
description: Структурата разделя приемането на временните файлове и изпращането към Incus контейнера в обработчика за качване на файлове в net/http на отделни операции, като използва буферизиран канал и пул от goroutine работници, с цел намаляване на забавянето на отговора на крайната точка.
language: bg
date: 2025-07-14T11:51:27.614782351Z
path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238
category: concurrency
---

# Общ преглед

Когато за първи път учим Go, често го правим, като реализираме сървър на бекенд. В такъв случай един често срещан пример е създаването на функционалност, която приема файлов поток през RestAPI и го качва на сървъра.
Сървърът net/http на езика Go по подразбиране обработва множество заявки едновременно, така че самото едновременно и масово качване на файлове няма проблем.
Но ако всички операции след приемането на потока се изпълняват синхронно, отговорът на крайната точка ще се забави.
Нека разгледаме техники, които да предотвратяват тази ситуация.

# Причина

Приемането на поток обикновено също отнема значително време, а при големи файлове една заявка може да се обработва няколко минути.
В такива случаи е важно да се обработват операциите след приемането възможно най-бързо.
Сценарият в този пример е следният: потокът се приема, запазва се като временен файл и след това се изпраща в контейнер.
Ако частта за изпращане на временния файл в контейнера се обработва чрез работен пул (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)
}
```

Опашката от задачи в този пример е реализирана просто. Тя има прост устройство, при което задачите, поставени в опашката, се извличат от канала.

По-долу е представен метод на работник за изпращане на файла в контейнера след качването му.
Текущият метод използва безкрайен цикъл, за да осигури добра отзивчивост и лесна реализация, но в зависимост от целите може да се добавят алгоритми.

```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)
Самият проект съдържа много допълнителни компоненти, затова за изучаването на самия работник е достатъчно само накратко да видите как функцията за инициализация на работника се извиква от main.go и да продължите нататък. Модулът съдържа и други видове работници, които можете да разгледате.