GoSudaVerkorting van de responstijd bij HTTP-bestandsuploads in Go: implementatie van een asynchrone workerwachtrij op basis van channels
Overzicht
Wanneer men Go voor het eerst leert, gebeurt het vaak dat men dit doet door een backendserver te implementeren. Laten we als voorbeeld een geval nemen waarin een RestAPI of vergelijkbaar systeem een bestandsstroom ontvangt en deze naar de server uploadt. De net/http-server van de Go-taal verwerkt standaard meerdere verzoeken tegelijkertijd, dus op het uploaden van gelijktijdig veel bestanden zelf is geen probleem. Als men echter alle bewerkingen na het ontvangen van de stroom synchroon uitvoert, wordt de respons van het eindpunt vertraagd. Laten we een techniek bekijken om deze situatie te voorkomen.
Oorzaak
Het ontvangen van een stroom kost over het algemeen ook veel tijd, en bij grote bestanden kan een enkel verzoek enkele minuten in beslag nemen. In zulke gevallen is het belangrijk om de bewerkingen na ontvangst zo snel mogelijk af te handelen. Het scenario van dit voorbeeld is: de stroom ontvangen, opslaan als tijdelijk bestand en vervolgens naar een container pushen. Als het pushen van het tijdelijke bestand naar de container hierbij wordt afgehandeld met een worker pool, kan de responsvertraging worden verkort.
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" // Host temporary directory
14
15// UploadTask holds data for asynchronous file push.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler processes file uploads. Saves to temp file, then queues for Incus push.
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 // Create unique temporary file path on host
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 // Create and copy request body to temporary file (synchronous)
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 // Enqueue task for asynchronous Incus push
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 immediate 202 Accepted response
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Bij het gedeelte dat is aangeduid met THIS PART, is het waarschijnlijk opgevallen dat hier een taak in de wachtrij wordt geplaatst.
Laten we nu eens kijken hoe de taakwachtrij werkt.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue initializes the in-memory task queue.
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 adds an UploadTask to the queue.
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 retrieves an UploadTask from the queue, blocking if empty.
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 returns current queue size.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
De als voorbeeld gegeven taakwachtrij is eenvoudig geïmplementeerd. Deze wachtrij heeft een eenvoudige structuur waarbij de taken in de wachtrij uit het kanaal worden gehaald.
Hieronder volgt de workermethode voor het pushen naar de container na het uploaden van het bestand. Voor een goede responsiviteit en eenvoudige implementatie is de huidige methode een oneindige lus, maar afhankelijk van het doel mag men er een algoritme aan toevoegen.
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 // Defer cleanup of the temporary file
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 // Process task with retries for transient Incus errors
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" { // Example permanent error
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}
Allereerst blijft deze functie in een lus draaien en probeert zij steeds taken uit de wachtrij op te halen. Vervolgens probeert zij binnen het bereik van de herpogingen de stap van het uploaden van het tijdelijke bestand naar de container uit te voeren, en niet de stap van stroom naar tijdelijk bestand.
Voordelen
Het voordeel van deze verwerkingswijze is dat, zolang de streamupload normaal verloopt, de vertraging van de daaropvolgende bewerkingen kan worden verminderd, en dat uitputting van bronnen door gelijktijdige containerbewerkingen kan worden voorkomen. Zoals in de huidige code te zien is, wordt het aantal containerbewerkingen dat tegelijk kan worden uitgevoerd beperkt door het aantal kanalen. Op deze manier hebben we een voorbeeld besproken dat op praktische wijze gebruik maakt van de parallelle verwerking van Go. Wilt u meer voorbeelden zien, dan kunt u de onderstaande links bezoeken.Module met voorbeelden Project dat de voorbeelden gebruikt Het project zelf bevat veel bijkomende componenten, dus voor het leren over de worker zelf volstaat het om kort te bekijken hoe de worker-init-functie vanuit main.go wordt aangeroepen en daarna verder te gaan. De module bevat ook andere vormen van workers, dus raadpleeg die gerust.