---
id: 8wel7k12347894gsdjkfnru23kl78989
author: Yunjin Lee
title: 'Skrócenie czasu odpowiedzi przy przesyłaniu plików HTTP w Go: implementacja asynchronicznej kolejki zadań roboczych opartej na kanałach'
description: Struktura ta rozdziela odbiór plików tymczasowych w handlerze przesyłania plików net/http oraz operacje transferu do kontenerów Incus na kanał buforowany i pulę workerów opartą na goroutines, dzięki czemu skraca się opóźnienie odpowiedzi endpointu.
language: pl
date: 2025-07-14T11:51:27.614782351Z
path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238
category: concurrency
---

# Przegląd

Ucząc się Go po raz pierwszy, często uczymy się, implementując serwer backendowy. Weźmy jako przykład sytuację, w której tworzymy przykład przesyłania pliku ze strumienia odebranego przez RestAPI lub podobne rozwiązanie na serwer.
Serwer net/http w języku Go domyślnie obsługuje wiele żądań jednocześnie, więc sama współbieżna, równoczesna wysyłka plików nie stanowi problemu.
Jeśli jednak wszystkie operacje następujące po odebraniu strumienia wykonamy synchronicznie, odpowiedź punktu końcowego ulegnie opóźnieniu.
Przyjrzyjmy się technikom, które pozwalają temu zapobiec.

# Przyczyna

Odebranie strumienia zazwyczaj również zajmuje sporo czasu, a w przypadku dużych plików pojedyncze żądanie może być przetwarzane przez kilka minut.
W takich przypadkach istotne jest, aby jak najszybciej przetwarzać operacje następujące po odbiorze.
Scenariusz tego przykładu polega na odebraniu strumienia, zapisaniu go do pliku tymczasowego, a następnie wysłaniu do kontenera.
Jeśli w tym miejscu część wysyłająca plik tymczasowy do kontenera zostanie obsłużona przez pulę workerów (worker pool), można skrócić opóźnienie odpowiedzi.

```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)
}
```

Jak zauważyłeś w miejscu oznaczonym jako THIS PART, zadanie jest wstawiane do kolejki.

Przyjrzyjmy się teraz temu, jak działa kolejka zadań.

```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)
}
```

Przykładowa kolejka zadań została zaimplementowana w prosty sposób. Ma ona prostą strukturę, w której zadania znajdujące się w kolejce są pobierane z kanału.

Poniżej znajduje się metoda workera, która po przesłaniu pliku wysyła go do kontenera.
Obecna metoda, dla zapewnienia dobrej responsywności i łatwej implementacji, działa w nieskończonej pętli, jednak w zależności od potrzeb można dodać do niej dodatkowe algorytmy.

```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
			}
		}
	}
}
```

Najpierw ta funkcja w nieskończonej pętli stara się pobierać zadania z kolejki. Następnie, w ramach zakresu ponowień, podejmuje próbę etapu wysyłki z pliku tymczasowego do kontenera, a nie ze strumienia do pliku tymczasowego.

# Zalety

Zaletą tego sposobu przetwarzania jest to, że **pod warunkiem, że przesyłanie strumieniowe przebiega prawidłowo**, można skrócić opóźnienie zadań przetwarzanych później oraz zapobiec wyczerpaniu zasobów spowodowanemu równoczesnymi operacjami na kontenerach. Jak widać w obecnym kodzie, liczba równocześnie wykonywanych operacji na kontenerach jest ograniczona pojemnością kanału. W ten praktyczny sposób poznaliśmy przykład, w którym można zastosować równoległe przetwarzanie w Go. Jeśli chcesz zobaczyć więcej przykładów, odwiedź poniższe linki.
[Moduł zawierający przykłady](https://github.com/gg582/linux_virt_unit)
[Projekt wykorzystujący przykład](https://github.com/gg582/incuspeed)
Sam projekt zawiera wiele elementów pomocniczych, dlatego do nauki samego workera wystarczy pobieżnie zapoznać się z tym, jak w main.go wywoływana jest funkcja init workera, i przejść dalej. Moduł zawiera również inne typy workerów, warto je zobaczyć.