GoSudaReducerea întârzierii răspunsului la încărcarea fișierelor HTTP în Go: implementarea unei cozi de lucru asincrone bazate pe canale
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ă.
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}
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.
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}
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.
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}
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 Proiect care folosește exemplul 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ță.