GoSudaMempersingkat Waktu Respons Unggah File HTTP di Go: Implementasi Antrean Worker Asinkron Berbasis Channel
Ikhtisar
Ketika mempelajari Go untuk pertama kali, sering kali kita belajar sambil mengimplementasikan server backend. Mari kita ambil contoh membuat contoh yang menerima aliran file dari RestAPI atau sejenisnya dan mengunggahnya ke server. Server net/http pada bahasa Go secara default memproses beberapa permintaan secara bersamaan, sehingga tidak ada masalah pada unggahan yang terjadi secara serentak dan bersamaan. Namun, jika semua operasi setelah penerimaan aliran diproses secara sinkron, respons endpoint akan tertunda. Mari kita pelajari teknik untuk mencegah situasi ini.
Penyebab
Menerima aliran umumnya juga membutuhkan waktu yang cukup lama, dan untuk file berukuran besar, satu permintaan dapat diproses selama beberapa menit. Dalam kasus seperti ini, penting untuk memproses operasi setelah penerimaan secepat mungkin. Skenario contoh ini adalah skenario menyimpan aliran yang diterima sebagai file sementara, lalu mendorongnya ke sebuah container. Pada saat ini, jika bagian yang mendorong file sementara ke container diproses dengan worker pool, keterlambatan respons dapat dipersingkat.
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" // Direktori sementara host
14
15// UploadTask menyimpan data untuk pendorongan file secara asinkron.
16type UploadTask struct {
17 HostTempFilePath string
18 ContainerName string
19 HostFilename string
20 ContainerDestinationPath string
21}
22
23// UploadHandler memproses unggahan file. Menyimpan ke file sementara, lalu memasukkannya ke antrean untuk pendorongan Incus.
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 // Membuat jalur file sementara yang unik di 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 // Membuat file sementara dan menyalin body permintaan ke dalamnya (sinkron)
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 // Memasukkan tugas ke antrean untuk pendorongan Incus secara asinkron
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 // Mengirim respons 202 Accepted secara langsung
83 wr.WriteHeader(http.StatusAccepted)
84 fmt.Fprintf(wr, "File '%s' queued for processing on container '%s'.\n", originalFilename, containerName)
85}
Jika Anda memperhatikan bagian yang diberi tanda THIS PART, Anda tentu menyadari bahwa tugas dimasukkan ke dalam antrean.
Sekarang mari kita lihat bagaimana antrean tugas bekerja.
1package file_upload
2
3import (
4 "log"
5 "sync"
6)
7
8var taskQueue chan UploadTask
9var once sync.Once
10
11// InitWorkQueue menginisialisasi antrean tugas dalam memori.
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 menambahkan UploadTask ke antrean.
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 mengambil UploadTask dari antrean, memblokir jika kosong.
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 mengembalikan ukuran antrean saat ini.
39func GetQueueLength() int {
40 if taskQueue == nil {
41 return 0
42 }
43 return len(taskQueue)
44}
Antrean tugas yang disediakan sebagai contoh diimplementasikan secara sederhana. Antrean tugas ini memiliki struktur sederhana yang mengeluarkan tugas-tugas yang berada di dalam antrean dari channel.
Berikut ini adalah metode worker untuk mendorong file ke container setelah unggahan file. Demi responsivitas yang baik dan kemudahan implementasi, metode saat ini menggunakan loop tak terbatas, namun algoritma dapat ditambahkan sesuai kebutuhan.
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 // Menjadwalkan pembersihan file sementara dengan defer
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 // Memproses tugas dengan percobaan ulang untuk kesalahan Incus yang bersifat sementara
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" { // Contoh kesalahan permanen
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}
Pertama, fungsi ini terus berputar dalam loop dan berusaha mengambil tugas dari antrean. Selanjutnya, dalam batas percobaan ulang, fungsi ini mencoba tahap unggahan dari file sementara ke container, bukan dari aliran ke file sementara.
Manfaat
Manfaat dari cara pemrosesan ini adalah, selama unggahan aliran berjalan normal, waktu tunggu untuk operasi yang diproses setelahnya dapat dikurangi, dan kehabisan sumber daya akibat operasi container yang terjadi secara bersamaan dapat dicegah. Seperti terlihat pada kode saat ini, jumlah operasi container yang dapat berjalan secara bersamaan dibatasi oleh kapasitas channel. Dengan demikian, kami telah mempelajari contoh yang secara praktis dapat memanfaatkan pemrosesan paralel di Go. Jika Anda ingin melihat lebih banyak contoh, silakan kunjungi tautan di bawah ini.Modul yang berisi contoh Proyek yang memanfaatkan contoh Proyek itu sendiri memiliki banyak komponen pendukung, jadi untuk mempelajari worker itu sendiri, cukup lihat secara singkat bagaimana fungsi init worker dipanggil dari main.go lalu lanjutkan. Modul ini juga berisi worker dengan bentuk lain, silakan dijadikan referensi.