--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Сокращение задержки ответа при загрузке файлов по HTTP в Go: реализация асинхронной очереди воркеров на основе каналов' description: Данная структура разделяет в обработчике загрузки файлов net/http приём временного файла и передачу в контейнер Incus на буферизованный канал и пул рабочих горутин, что позволяет сократить задержку ответа конечной точки. language: ru date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # Обзор Когда начинаешь изучать Go, часто учишься, реализуя серверную часть бэкенда. В таком случае, например, можно создать пример, в котором через RestAPI и т. п. принимается файловый поток и загружается на сервер. HTTP-сервер net/http в Go по умолчанию обрабатывает несколько запросов одновременно, поэтому сама по себе параллельная загрузка нескольких файлов не представляет проблемы. Однако если все действия после приёма потока выполнять синхронно, ответ конечной точки будет задерживаться. Рассмотрим технику, позволяющую этого избежать. # Причина Приём потока, как правило, также занимает много времени, а в случае больших файлов один запрос может обрабатываться несколько минут. В таких случаях важно как можно быстрее приступить к действиям после приёма данных. Сценарий этого примера заключается в том, чтобы принять поток, сохранить его во временный файл и отправить в контейнер. Если обрабатывать отправку временного файла в контейнер с помощью пула воркеров, можно сократить задержку ответа. ```go package file_upload import ( "fmt" "io" "log" "net/http" "os" "path/filepath" "time" ) const uploadTempDir = "/tmp/incus_uploads" // Временный каталог на хосте // UploadTask хранит данные для асинхронной отправки файла. type UploadTask struct { HostTempFilePath string ContainerName string HostFilename string ContainerDestinationPath string } // UploadHandler обрабатывает загрузку файлов. Сохраняет во временный файл, затем ставит в очередь для отправки в Incus. 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 } // Создание уникального пути временного файла на хосте 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 } // Создание временного файла и копирование тела запроса в него (синхронно) 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) // Постановка задачи в очередь для асинхронной отправки в Incus 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) // Немедленная отправка ответа 202 Accepted 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 инициализирует очередь задач в памяти. func InitWorkQueue() { once.Do(func() { taskQueue = make(chan UploadTask, 100) log.Println("Upload Info: Work queue initialized.") }) } // EnqueueTask добавляет UploadTask в очередь. 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 извлекает UploadTask из очереди, блокируясь, если она пуста. 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 возвращает текущий размер очереди. 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 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) // Обработка задачи с повторными попытками при временных ошибках Incus for i := 0; i <= MaxRetries; i++ { err := processUploadTask(task) //отдельная задача загрузки if err == nil { log.Printf("SUCCESS: Worker: Task completed for %s.", task.ContainerName) break } isTransient := true if err.Error() == "incus: container not found" { // Пример постоянной ошибки 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 вызывается функция инициализации воркера. В модуле также есть другие виды воркеров, см. их, пожалуйста.