--- id: 8wel7k12347894gsdjkfnru23kl78989 author: Yunjin Lee title: 'Verkürzung der Antwortzeit beim Datei-Upload per Go HTTP: Implementierung einer kanalbasierten asynchronen Worker-Queue' description: Die Struktur trennt in dem Datei-Upload-Handler von net/http das Empfangen temporärer Dateien und die Übertragung in Incus-Container durch einen gepufferten Channel und einen Worker-Pool aus Goroutinen, wodurch die Antwortlatenz des Endpunkts reduziert wird. language: de date: 2025-07-14T11:51:27.614782351Z path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238 category: concurrency --- # Übersicht Wenn man Go zum ersten Mal lernt, geschieht es häufig, dass man dabei einen Backend-Server implementiert. Nehmen wir als Beispiel den Fall, dass man ein Beispiel erstellt, bei dem ein Dateistream über eine REST-API empfangen und auf den Server hochgeladen wird. Der net/http-Server von Go verarbeitet standardmäßig mehrere Anfragen gleichzeitig, daher gibt es beim gleichzeitigen, massenhaften Hochladen an sich kein Problem. Wenn jedoch sämtliche Vorgänge nach dem Empfang des Streams synchron verarbeitet werden, verzögert sich die Antwort des Endpunkts. Lassen Sie uns Techniken betrachten, mit denen diese Situation vermieden werden kann. # Ursache Auch das Empfangen eines Streams benötigt im Allgemeinen viel Zeit; bei großen Dateien kann eine einzelne Anfrage mehrere Minuten in Anspruch nehmen. In solchen Fällen ist es wichtig, die Vorgänge nach dem Empfang möglichst zügig zu verarbeiten. Das Beispielszenario besteht darin, den Stream zu empfangen, als temporäre Datei zu speichern und anschließend in einen Container zu übertragen. Wenn dabei der Teil, der die temporäre Datei in den Container überträgt, über einen Worker-Pool verarbeitet wird, lässt sich die Antwortverzögerung verkürzen. ```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) } ``` Den Abschnitt, den wir mit THIS PART markiert haben, haben Sie sicherlich bemerkt: Dort wird die Aufgabe in eine Warteschlange eingefügt. Schauen wir uns nun an, wie die Aufgabenwarteschlange funktioniert. ```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) } ``` Die als Beispiel bereitgestellte Aufgabenwarteschlange ist einfach implementiert. Sie besitzt eine schlichte Struktur, bei der die in der Warteschlange befindlichen Aufgaben aus dem Channel entnommen werden. Im Folgenden finden Sie die Worker-Methode, die nach dem Datei-Upload die Übertragung in den Container übernimmt. Die aktuelle Methode ist für gute Reaktionsfähigkeit und einfache Implementierung als Endlosschleife ausgelegt; je nach Zweck kann jedoch auch ein zusätzlicher Algorithmus ergänzt werden. ```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 } } } } ``` Zunächst versucht diese Funktion in einer Schleife, Tasks aus der Warteschlange abzurufen. Anschließend wird innerhalb des Wiederholungsbereichs der Upload-Schritt ausgeführt, der nicht vom Stream zur temporären Datei, sondern von der temporären Datei zum Container erfolgt. # Vorteile Der Vorteil dieser Verarbeitungsweise liegt darin, dass **solange der Stream-Upload erfolgreich verläuft** sich die Latenz der nachfolgenden Verarbeitung verringern lässt und eine Erschöpfung der Ressourcen durch gleichzeitige Container-Operationen vermieden werden kann. Wie im aktuellen Code ersichtlich ist, wird die Anzahl der gleichzeitig ausführbaren Container-Operationen durch die Kapazität des Channels begrenzt. Auf diese Weise haben wir ein Beispiel kennengelernt, das die Parallelverarbeitung in Go praxisnah nutzbar macht. Wenn Sie weitere Beispiele sehen möchten, besuchen Sie bitte die folgenden Links. [Modul mit Beispielen](https://github.com/gg582/linux_virt_unit) [Projekt unter Verwendung des Beispiels](https://github.com/gg582/incuspeed) Da das Projekt selbst viele Begleitkomponenten enthält, genügt es für das Erlernen des Workers selbst, kurz zu betrachten, wie die Worker-Init-Funktion in main.go aufgerufen wird, und dann weiterzugehen. Das Modul enthält außerdem weitere Arten von Workern; bitte schauen Sie sich diese ebenfalls an.