--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Reducción de la latencia de respuesta en la carga de archivos HTTP en Go: implementación de una cola de trabajadores asíncronos basada en canales' description: Esta estructura separa, en el manejador de carga de archivos de net/http, la recepción de archivos temporales y las operaciones de transferencia a contenedores Incus mediante un canal con búfer y un pool de workers basado en goroutines, con el fin de reducir la latencia de respuesta del endpoint. language: es date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # Descripción general Cuando se aprende Go por primera vez, con frecuencia se hace mediante la implementación de un servidor backend. Consideremos un ejemplo en el que se crea un caso práctico para recibir un flujo de archivos a través de una API REST y subirlo al servidor. El servidor net/http del lenguaje Go procesa múltiples solicitudes de manera simultánea por defecto, por lo que las subidas concurrentes en sí mismas no representan un problema. Sin embargo, si todas las operaciones posteriores a la recepción del flujo se procesan de forma sincrónica, la respuesta del endpoint se retrasará. Examinemos las técnicas disponibles para prevenir esta situación. # Causa La recepción del flujo también suele demandar un tiempo considerable; en el caso de archivos grandes, una sola solicitud puede tardar varios minutos en procesarse. En tales circunstancias, resulta crucial procesar las operaciones posteriores a la recepción con la mayor rapidez posible. Este escenario de ejemplo consiste en recibir el flujo, guardarlo como un archivo temporal y posteriormente enviarlo a un contenedor. Si la parte correspondiente al envío del archivo temporal al contenedor se gestiona mediante un grupo de trabajadores (worker pool), es posible acortar el retraso en la respuesta. ```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) } ``` Al observar la sección marcada como THIS PART aquí, probablemente habrá notado que las tareas se insertan en una cola. Ahora pasemos a examinar cómo opera la cola de tareas. ```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) } ``` La cola de tareas proporcionada como ejemplo cuenta con una implementación sencilla. Dicha cola posee una estructura simple que extrae del canal las tareas almacenadas en ella. A continuación se muestra el método del trabajador destinado a realizar el envío al contenedor tras la subida del archivo. El método actual utiliza un bucle infinito para lograr una buena capacidad de respuesta y una implementación simple, aunque es perfectamente admisible añadir algoritmos adicionales según los requerimientos específicos. ```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 } } } } ``` En primer lugar, esta función ejecuta un bucle continuo mientras intenta obtener tareas desde la cola. A continuación, dentro del rango de reintentos establecido, procede con la fase de carga desde el archivo temporal hacia el contenedor, en lugar de utilizar el flujo directamente. # Ventajas La ventaja de este enfoque de procesamiento radica en que, **siempre que la subida del flujo sea correcta**, es posible reducir el tiempo de latencia de las operaciones posteriores y prevenir el agotamiento de recursos provocado por tareas simultáneas en los contenedores. Tal como se observa en el código actual, la cantidad de operaciones en contenedores que se pueden ejecutar de manera concurrente está limitada por la capacidad del canal. De este modo, hemos examinado un ejemplo práctico de cómo aplicar el procesamiento en paralelo de Go. Si desea consultar más ejemplos, visite los siguientes enlaces: [Módulo con ejemplos incluidos](https://github.com/gg582/linux_virt_unit) [Proyecto que utiliza los ejemplos](https://github.com/gg582/incuspeed) Dado que el proyecto en sí contiene numerosos componentes secundarios, para comprender el funcionamiento del trabajador basta con revisar brevemente cómo se invoca la función de inicialización del trabajador dentro de main.go. Tenga en cuenta que el módulo también incluye otras variantes de trabajadores.