GoSudaReduktion af svartid ved HTTP-filupload i Go: implementering af en asynkron worker-kø baseret på kanaler
Oversigt
Når man første gang lærer Go, sker det ofte, at man lærer ved at implementere en backend-server. Lad os her tage som eksempel et tilfælde, hvor man laver et eksempel, der modtager en filstrøm via en RestAPI o.lign. og uploader den til serveren. Go-sprogets net/http-server håndterer som standard flere forespørgsler samtidigt, så der er i sig selv ingen problemer med samtidige, mange-til-én-uploads. Men hvis alle handlinger efter modtagelsen af strømmen udføres synkront, forsinkes endpointets svar. Lad os se på teknikker til at forhindre denne situation.
Årsag
Det tager som regel også lang tid at modtage en strøm, og for store filer kan en enkelt forespørgsel tage flere minutter at behandle. I et sådant tilfælde er det vigtigt at behandle handlingerne efter modtagelsen så hurtigt som muligt. Scenariet i dette eksempel er at gemme strømmen som en midlertidig fil og derefter skubbe den til en container. Hvis delen, hvor den midlertidige fil skubbes til containeren, behandles med en worker pool, kan svartiden forkortes.
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" // Midlertidig mappe på værten
14
15// UploadTask holder data til asynkron filoverførsel.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler behandler filuploads. Gemmer til midlertidig fil og sætter derefter i kø til Incus-overførsel.
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 // Opret unik midlertidig filsti på værten
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 // Opret midlertidig fil og kopier request body til den (synkront)
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 // Sæt opgave i kø til asynkron Incus-overførsel
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 // Send øjeblikkeligt 202 Accepted-svar
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Når man ser på den del, der er markeret med THIS PART, vil du have bemærket, at opgaven indsættes i en kø.
Lad os nu se, hvordan opgavekøen fungerer.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue initialiserer den in-memory opgavekø.
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 tilføjer en UploadTask til køen.
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 henter en UploadTask fra køen og blokerer, hvis køen er tom.
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 returnerer den aktuelle køstørrelse.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
Den eksempelvis angivne opgavekø er implementeret på en enkel måde. Denne opgavekø har en simpel struktur, der tager opgaverne ud af kanalen i den rækkefølge, de er blevet sat i kø.
Nedenfor er worker-metoden til at skubbe filen til containeren efter filoverførslen. Af hensyn til god responsivitet og nem implementering er den nuværende metode en uendelig løkke, men der kan tilføjes algoritmer afhængigt af formålet.
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 // Udskyd oprydning af den midlertidige fil
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 // Behandl opgaven med genforsøg ved midlertidige Incus-fejl
16 for i := 0; i <= MaxRetries; i++ {
17 err := processUploadTask(task) //separate upload 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" { // Eksempel på en permanent fejl
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}
Først kører denne funktion en løkke, der gentagne gange forsøger at hente opgaver fra køen. Derefter forsøger den, inden for rammerne af genforsøgene, at udføre uploadtrinnet fra den midlertidige fil til containeren, og ikke fra strømmen til den midlertidige fil.
Fordele
Fordelen ved denne behandlingsmetode er, at så længe strømuploadet fungerer korrekt, kan forsinkelsen for de efterfølgende handlinger reduceres, og ressourceudtømning på grund af samtidige containeroperationer kan forebygges. Som det fremgår af den nuværende kode, er antallet af containeroperationer, der kan køre samtidigt, begrænset af kanalens kapacitet. Sådan har vi gennemgået et eksempel, der praktisk anvender Gos parallelle processering. Hvis du ønsker at se flere eksempler, så besøg linkene nedenfor.Modul med eksempler Projekt, der anvender eksemplet Projektet i sig selv indeholder mange sekundære komponenter, så for at lære om selve workeren er det tilstrækkeligt kort at se, hvordan worker-init-funktionen kaldes i main.go, og så gå videre. Modulet indeholder også andre former for workers, så se gerne dem også.