--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Go HTTP fájlfeltöltési válaszidő csökkentése: csatorna alapú aszinkron munkasor implementálása' description: A net/http fájlfeltöltő kezelőben a Incus konténerbe történő átvitel és az ideiglenes fájlok fogadása buffered channel és goroutine worker pool segítségével kerül szétválasztásra, ezáltal csökkentve a végpont válaszidejét. language: hu date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # Áttekintés Amikor az ember először tanulja a Go nyelvet, gyakran a backend szerver implementálása közben sajátítja el a nyelvet. Ilyenkor vegyünk példának egy olyan példaprogramot, amely RestAPI stb. segítségével fájlfolyamot fogad, majd feltölti a szerverre. A Go nyelv net/http szervere alapértelmezés szerint több kérést is párhuzamosan kezel, ezért maga a párhuzamos, egyidejű feltöltés nem okoz problémát. Ha azonban a folyam fogadása utáni összes műveletet szinkronban dolgozzuk fel, az végpont válaszát késlelteti. Nézzük meg, milyen technikával előzhetjük meg ezt a helyzetet. # Ok A folyam fogadása is általában hosszú időt vesz igénybe, nagy fájlok esetén pedig egyetlen kérés feldolgozása több percig is tarthat. Ilyen esetekben kulcsfontosságú, hogy a fogadás utáni műveleteket a lehető leggyorsabban kezdjük el feldolgozni. Ennek a példaforgatókönyvnek az a lényege, hogy a folyamot fogadás után ideiglenes fájlba mentjük, majd egy konténerbe töltjük fel. Ha a konténerbe történő ideiglenes fájl-feltöltést munkavégző-készlettel (worker pool) oldjuk meg, a válaszidő csökkenthető. ```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) } ``` Ha a THIS PART megjelöléssel ellátott részt nézzük, bizonyára észrevette, hogy ott a feladat sorba állításáról van szó. Most nézzük meg, hogyan működik a feladatsor. ```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) } ``` A példaként bemutatott feladatsor egyszerűen van megvalósítva. Ez a feladatsor egy egyszerű szerkezetű, amely a sorban álló feladatokat a csatornából veszi ki. Az alábbi a fájlfeltöltés utáni, konténerbe történő feltöltést végző worker metódus. A jelenlegi metódus a jó reagálóképesség és az egyszerű megvalósítás érdekében végtelen ciklust használ, de a céltól függően algoritmusok is hozzáadhatók. ```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 } } } } ``` Először is ez a függvény folyamatosan ciklusban fut, és megpróbálja a feladatokat a sorból átvenni. Ezt követően az újrapróbálkozási kereten belül megkísérli a feltöltési lépést, amely már nem a folyamból az ideiglenes fájlba, hanem az ideiglenes fájlból a konténerbe történik. # Előnyök Ennek a feldolgozási módnak az előnye, hogy **ha csak a folyamfeltöltés sikeres**, a későbbi feldolgozás késleltetése csökkenthető, és a egyidejű, nagy számú konténerművelet okozta erőforrás-kimerülés is megelőzhető. A jelenlegi kódból látszik, hogy egyszerre folyamatban lévő konténerműveletek számát a csatorna mérete korlátozza. Így a Go párhuzamos feldolgozását gyakorlatiasan alkalmazható példán keresztül ismertük meg. Ha további példákat szeretne látni, kérjük, keresse fel az alábbi linkeket. [Példákat tartalmazó modul](https://github.com/gg582/linux_virt_unit) [Példát felhasználó projekt](https://github.com/gg582/incuspeed) Maga a projekt sok kiegészítő komponenst tartalmaz, ezért a worker önmagának tanulmányozásához elegendő, ha a main.go fájlban csak röviden megnézi, hogyan hívjuk meg a worker init függvényét. A modul más típusú workereket is tartalmaz, amelyeket szintén érdemes megtekinteni.