--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Reducerea întârzierii răspunsului la încărcarea fișierelor HTTP în Go: implementarea unei cozi de lucru asincrone bazate pe canale' description: Structura separă recepția fișierelor temporare și operațiunile de transfer către containerele Incus din handler-ul de încărcare a fișierelor din net/http, prin utilizarea unui canal tampon (buffered channel) și a unui pool de workeri bazat pe goroutine, în scopul reducerii latenței de răspuns a endpoint-ului. language: ro date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # Prezentare generală Atunci când învață Go pentru prima dată, mulți oameni o fac implementând un server backend. În acest context, să luăm ca exemplu crearea unui exemplu în care un RestAPI sau similar primește un flux de fișier și îl încarcă pe server. Serverul net/http din limbajul Go procesează, în mod implicit, mai multe cereri simultan, așa că încărcarea concurentă a mai multor fișiere, în sine, nu prezintă probleme. Totuși, dacă toate operațiunile de după primirea fluxului sunt procesate sincron, răspunsul endpoint-ului va fi întârziat. Să analizăm tehnica folosită pentru a preveni această situație. # Cauza Primirea unui flux durează, de regulă, mult timp, iar în cazul fișierelor mari, o singură cerere poate fi procesată timp de mai multe minute. În astfel de situații, este important să procesăm rapid, oricât de puțin, operațiunile ulterioare primirii. Scenariul acestui exemplu constă în primirea fluxului, salvarea sa ca fișier temporar și apoi trimiterea (push) către un container. Dacă partea de trimitere a fișierului temporar către container este gestionată printr-un worker pool, întârzierea răspunsului poate fi redusă. ```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) } ``` Observând porțiunea marcată cu THIS PART, probabil ați remarcat că aceasta introduce o operațiune de inserare a sarcinii în coadă. Să vedem acum cum funcționează coada de sarcini. ```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) } ``` Coada de sarcini furnizată ca exemplu este implementată în mod simplu. Această coadă are o structură elementară, care extrage sarcinile aflate în coadă din canal. Mai jos se află metoda worker pentru trimiterea fișierelor în container după încărcarea acestora. Metoda curentă rulează într-o buclă infinită, pentru un răspuns bun și o implementare simplă, dar se pot adăuga algoritmi în funcție de scop. ```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 } } } } ``` Mai întâi, această funcție rulează continuu într-o buclă și încearcă să preia sarcinile din coadă. Apoi, în limitele numărului de reîncercări, încearcă etapa de încărcare de la fișierul temporar către container, nu de la flux la fișierul temporar. # Beneficii Beneficiul acestei abordări este că, **atâta timp cât încărcarea fluxului reușește**, se poate reduce întârzierea operațiunilor procesate ulterior și se poate preveni epuizarea resurselor cauzată de operațiunile concurente pe containere. Așa cum se observă în codul actual, numărul operațiunilor pe container care pot rula simultan este limitat de capacitatea canalului. Astfel, am analizat un exemplu practic de utilizare a paralelismului din Go. Dacă doriți să vedeți mai multe exemple, vă rugăm să vizitați linkurile de mai jos. [Modul cu exemple](https://github.com/gg582/linux_virt_unit) [Proiect care folosește exemplul](https://github.com/gg582/incuspeed) Proiectul în sine conține multe componente auxiliare, așa că, pentru studiul worker-ului în sine, este suficient să consultați pe scurt modul în care funcția init a worker-ului este apelată din main.go și apoi să treceți mai departe. Modulul include și alte tipuri de worker, despre care vă rugăm să aflați din materialul de referință.