GoSudaСокращение задержки ответа при загрузке файлов по HTTP в Go: реализация асинхронной очереди воркеров на основе каналов
Обзор
Когда начинаешь изучать Go, часто учишься, реализуя серверную часть бэкенда. В таком случае, например, можно создать пример, в котором через RestAPI и т. п. принимается файловый поток и загружается на сервер. HTTP-сервер net/http в Go по умолчанию обрабатывает несколько запросов одновременно, поэтому сама по себе параллельная загрузка нескольких файлов не представляет проблемы. Однако если все действия после приёма потока выполнять синхронно, ответ конечной точки будет задерживаться. Рассмотрим технику, позволяющую этого избежать.
Причина
Приём потока, как правило, также занимает много времени, а в случае больших файлов один запрос может обрабатываться несколько минут. В таких случаях важно как можно быстрее приступить к действиям после приёма данных. Сценарий этого примера заключается в том, чтобы принять поток, сохранить его во временный файл и отправить в контейнер. Если обрабатывать отправку временного файла в контейнер с помощью пула воркеров, можно сократить задержку ответа.
1package file_upload
2
3import (
4 "fmt"
5 "io"
6 "log"
7 "net/http"
8 "os"
9 "path/filepath"
10 "time"
11)
12
13const uploadTempDir = "/tmp/incus_uploads" // Временный каталог на хосте
14
15// UploadTask хранит данные для асинхронной отправки файла.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler обрабатывает загрузку файлов. Сохраняет во временный файл, затем ставит в очередь для отправки в Incus.
24func UploadHandler(wr http.ResponseWriter, req *http.Request) {
25 if req.Method != http.MethodPost {
26 http.Error(wr, "POST method required.", http.StatusMethodNotAllowed)
27 return
28 }
29 originalFilePath := req.Header.Get("X-File-Path")
30 originalFilename := filepath.Base(req.Header.Get("X-Host-Path"))
31 containerName := req.Header.Get("X-Container-Name")
32 if originalFilePath == "" || containerName == "" {
33 http.Error(wr, "Missing X-File-Path or X-Container-Name header.", http.StatusBadRequest)
34 return
35 }
36
37 cleanContainerDestPath := filepath.Clean(originalFilePath)
38 if !filepath.IsAbs(cleanContainerDestPath) {
39 http.Error(wr, "File path must be absolute.", http.StatusBadRequest)
40 return
41 }
42
43 // Создание уникального пути временного файла на хосте
44 tempFileName := fmt.Sprintf("%d-%s", time.Now().UnixNano(), filepath.Base(originalFilePath))
45 hostTempFilePath := filepath.Join(uploadTempDir, tempFileName)
46
47 if err := os.MkdirAll(uploadTempDir, 0755); err != nil {
48 log.Printf("ERROR: Failed to create temp upload directory: %v", err)
49 http.Error(wr, "Server error.", http.StatusInternalServerError)
50 return
51 }
52
53 // Создание временного файла и копирование тела запроса в него (синхронно)
54 outFile, err := os.Create(hostTempFilePath)
55 if err != nil {
56 log.Printf("ERROR: Failed to create temporary file: %v", err)
57 http.Error(wr, "Server error.", http.StatusInternalServerError)
58 return
59 }
60 defer outFile.Close()
61
62 bytesWritten, err := io.Copy(outFile, req.Body)
63 if err != nil {
64 outFile.Close()
65 os.Remove(hostTempFilePath)
66 log.Printf("ERROR: Failed to copy request body to temp file: %v", err)
67 http.Error(wr, "File transfer failed.", http.StatusInternalServerError)
68 return
69 }
70 log.Printf("Upload Info: Received %d bytes, saved to %s.", bytesWritten, hostTempFilePath)
71
72 // Постановка задачи в очередь для асинхронной отправки в Incus
73 task := UploadTask{
74 HostTempFilePath: hostTempFilePath,
75 ContainerName: containerName,
76 HostFilename : originalFilename,
77 ContainerDestinationPath: cleanContainerDestPath,
78 }
79 EnqueueTask(task) //THIS PART
80 log.Printf("Upload Info: Task enqueued for %s to %s.", originalFilePath, containerName)
81
82 // Немедленная отправка ответа 202 Accepted
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Если вы обратили внимание на место, отмеченное как THIS PART, то заметили, что здесь задача помещается в очередь.
Теперь посмотрим, как работает очередь задач.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue инициализирует очередь задач в памяти.
12func InitWorkQueue() {
13 once.Do(func() {
14 taskQueue = make(chan UploadTask, 100)
15 log.Println("Upload Info: Work queue initialized.")
16 })
17}
18
19// EnqueueTask добавляет UploadTask в очередь.
20func EnqueueTask(task UploadTask) {
21 if taskQueue == nil {
22 log.Fatal("ERROR: Task queue not initialized.")
23 }
24 taskQueue <- task
25 log.Printf("Upload Info: Queue: Task enqueued. Size: %d", len(taskQueue))
26}
27
28// DequeueTask извлекает UploadTask из очереди, блокируясь, если она пуста.
29func DequeueTask() UploadTask {
30 if taskQueue == nil {
31 log.Fatal("ERROR: Task queue not initialized.")
32 }
33 task := <-taskQueue
34 log.Printf("Upload Info: Queue: Task dequeued. Size: %d", len(taskQueue))
35 return task
36}
37
38// GetQueueLength возвращает текущий размер очереди.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
Пример очереди задач реализован простым образом. Эта очередь имеет простую структуру, которая извлекает задачи из канала по мере их постановки в очередь.
Ниже приведён метод воркера, который отправляет файл в контейнер после загрузки. Для хорошей отзывчивости и простоты реализации текущий метод использует бесконечный цикл, однако в зависимости от назначения можно добавить и другие алгоритмы.
1func StartFilePushWorker() {
2 for {
3 task := DequeueTask()
4 log.Printf("Upload Info: Worker processing task for %s from %s.", task.ContainerName, task.HostFilename)
5
6 // Отложенная очистка временного файла
7 defer func(filePath string) {
8 if err := os.Remove(filePath); err != nil {
9 log.Printf("ERROR: Worker: Failed to remove temp file '%s': %v", filePath, err)
10 } else {
11 log.Printf("Upload Info: Worker: Cleaned up temp file: %s", filePath)
12 }
13 }(task.HostTempFilePath)
14
15 // Обработка задачи с повторными попытками при временных ошибках Incus
16 for i := 0; i <= MaxRetries; i++ {
17 err := processUploadTask(task) //отдельная задача загрузки
18 if err == nil {
19 log.Printf("SUCCESS: Worker: Task completed for %s.", task.ContainerName)
20 break
21 }
22
23 isTransient := true
24 if err.Error() == "incus: container not found" { // Пример постоянной ошибки
25 isTransient = false
26 }
27
28 if isTransient && i < MaxRetries {
29 log.Printf("WARNING: Worker: Task failed for %s (attempt %d/%d): %v. Retrying.",
30 task.ContainerName, i+1, MaxRetries, err)
31 time.Sleep(RetryDelay)
32 } else {
33 log.Printf("ERROR: Worker: Task permanently failed for %s after %d attempts: %v.",
34 task.ContainerName, i+1, err)
35 break
36 }
37 }
38 }
39}
Сначала эта функция в бесконечном цикле пытается получить задачу из очереди. Затем, в пределах допустимого количества повторных попыток, она выполняет этап загрузки из временного файла в контейнер (а не из потока во временный файл).
Преимущества
Преимущество такого подхода заключается в том, что при нормальной загрузке потока можно сократить задержку последующей обработки и предотвратить исчерпание ресурсов из-за одновременных операций с контейнерами. Как видно из текущего кода, число одновременно выполняемых операций с контейнерами ограничено ёмкостью канала. Таким образом, мы рассмотрели пример, который практически демонстрирует использование параллельной обработки в Go. Если вы хотите увидеть больше примеров, перейдите по ссылкам ниже.Модуль с примерами Проект, использующий примеры В проекте много вспомогательных компонентов, поэтому для изучения самого воркера достаточно кратко посмотреть, как в main.go вызывается функция инициализации воркера. В модуле также есть другие виды воркеров, см. их, пожалуйста.