---
id: 8wel7k12347894gsdjkfnru23kl78989
author: Yunjin Lee
title: Go HTTP ファイルアップロード応答遅延の短縮：チャネルベース非同期ワーカーキューの実装
description: net/http のファイルアップロードハンドラーにおいて、一時ファイルの受信と Incus コンテナへの転送処理を、バッファ付きチャネルとゴルーチンのワーカープールによって分離し、エンドポイントの応答遅延を低減する構造です。
language: ja
date: 2025-07-14T11:51:27.614782351Z
path: /blog/posts/go-handle-file-upload-with-concurrency-s844ce238
category: concurrency
---

# 概要

Goを初めて学ぶ際、バックエンドサーバーを実装しながら学ぶケースがよくあります。この際、RestAPIなどでファイルストリームを受け取り、サーバーにアップロードする例を作成する場合を例に挙げてみましょう。
Go言語のnet/httpサーバーは基本的に複数のリクエストを同時に処理するため、同時多発的なアップロード自体には問題がありません。
しかし、ストリーム受信後の動作をすべて同期的に処理すると、エンドポイントの応答が遅延します。
このような状況を防ぐための手法について見ていきましょう。

# 原因

ストリームの受信にも一般的に長い時間がかかり、大きなファイルの場合、単一のリクエストが数分間処理されることもあります。
このような場合、受信後の動作をできるだけ迅速に処理することが重要です。
この例のシナリオは、ストリームを受信して一時ファイルとして保存し、コンテナにプッシュするシナリオです。
この際、コンテナへの一時ファイルのプッシュ部分をワーカープールで処理すれば、応答遅延を短縮できます。

```go
package file_upload

import (
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"time"
)

const uploadTempDir = "/tmp/incus_uploads" // ホストの一時ディレクトリ

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

ここでTHIS PARTと記した部分を見ると、タスクをキューに挿入していることにお気づきだと思います。

では、タスクキューがどのように動作するかを見てみましょう。

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

例として提示したタスクキューは単純に実装されています。このタスクキューは、キューに入っているタスクをチャネルから取り出すという単純な構造を持っています。

以下は、ファイルのアップロード後にコンテナへプッシュするためのワーカーメソッドです。
現在のメソッドは、良好な応答性と容易な実装のために無限ループを用いていますが、用途に応じてアルゴリズムを追加しても差し支えありません。

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

まず、この関数は繰り返しループを回りながら、キューからタスクを取得しようと試みます。その後、再試行の範囲内で、ストリームから一時ファイルへの処理ではなく、一時ファイルからコンテナへのアップロード段階を試みます。

# 利点

この処理方式の利点は、**ストリームのアップロードさえ正常であれば**、その後に処理される作業の遅延時間を短縮でき、同時多発的なコンテナ作業によるリソースの枯渇を予防できることです。現在のコードに見られるように、同時に進行可能なコンテナ作業はチャネルの容量によって制限されます。このように、実用的にGoの並列処理を活用できる例を確認しました。より多くの例を見たい場合は、以下のリンクを訪問してください。
[例を含むモジュール](https://github.com/gg582/linux_virt_unit)
[例を活用したプロジェクト](https://github.com/gg582/incuspeed)
プロジェクト自体には付随的な構成要素が多いため、ワーカー自体についての学習は、main.goでワーカーのinit関数をどのように呼び出しているかを簡単に確認する程度で済ませれば十分です。モジュールには別の形態のワーカーも含まれているので、ぜひ参考にしてください。