[Go 심화] [#1] 고루틴 풀 구현 및 관리

package main import ( "fmt" "sync" "time" ) // Task는 실행할 작업을 나타내는 함수 타입입니다. type Task func() error // Pool은 고루틴 풀을 나타내는 구조체입니다. type Pool struct { Tasks chan Task NumWorkers int WorkersDone sync.WaitGroup Quit chan bool } // NewPool은 새로운 고루틴 풀을 생성합니다. func NewPool(numWorkers int, queueSize int) *Pool { pool := &Pool{ Tasks: make(chan Task, queueSize), NumWorkers: numWorkers, Quit: make(chan bool), } // 워커 고루틴 시작 pool.WorkersDone.Add(numWorkers) for i := 0; i < numWorkers; i++ { go pool.worker(i) } return pool } // worker는 작업 큐에서 작업을 가져와 실행하는 고루틴입니다. func (p *Pool) worker(id int) { defer p.WorkersDone.Done() for { select { case task, ok := <-p.Tasks: if !ok { fmt.Printf("Worker %d shutting down\n", id) return } // 작업 실행 err := task() if err != nil { fmt.Printf("Worker %d encountered error: %v\n", id, err) } case <-p.Quit: fmt.Printf("Worker %d received quit signal\n", id) return } } } // AddTask는 작업을 풀에 추가합니다. func (p *Pool) AddTask(task Task) { p.Tasks <- task } // Close는 모든 작업이 완료될 때까지 기다린 후 풀을 종료합니다. func (p *Pool) Close() { close(p.Tasks) p.WorkersDone.Wait() } // Stop은 즉시 모든 워커를 중지시킵니다. func (p *Pool) Stop() { close(p.Quit) p.WorkersDone.Wait() } func main() { // 4개의 워커와 10개의 작업 큐 크기를 가진 풀 생성 pool := NewPool(4, 10) // 10개의 작업 추가 for i := 0; i < 10; i++ { taskID := i pool.AddTask(func() error { fmt.Printf("Task %d started\n", taskID) // 작업 시뮬레이션 time.Sleep(time.Second) fmt.Printf("Task %d completed\n", taskID) return nil }) } // 모든 작업이 완료될 때까지 대기 후 풀 종료 pool.Close() fmt.Println("All tasks completed") }

package main import ( "fmt" "log" "net/http" "sync" "time" ) // RequestPool은 HTTP 요청을 처리하기 위한 고루틴 풀입니다. type RequestPool struct { Tasks chan func() Workers int WorkerDone sync.WaitGroup } // NewRequestPool은 새로운 요청 처리 풀을 생성합니다. func NewRequestPool(workers int) *RequestPool { pool := &RequestPool{ Tasks: make(chan func(), 100), Workers: workers, } pool.WorkerDone.Add(workers) for i := 0; i < workers; i++ { go func(workerID int) { defer pool.WorkerDone.Done() for task := range pool.Tasks { task() } }(i) } return pool } // Close는 풀을 종료합니다. func (p *RequestPool) Close() { close(p.Tasks) p.WorkerDone.Wait() } func main() { // 8개의 워커를 가진 요청 풀 생성 pool := NewRequestPool(8) defer pool.Close() // HTTP 핸들러 설정 http.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { // 요청 정보 캡처 reqPath := r.URL.Path // 응답 채널 생성 responseCh := make(chan string, 1) // 작업을 풀에 추가 pool.Tasks <- func() { // 작업 처리 시뮬레이션 time.Sleep(100 * time.Millisecond) result := fmt.Sprintf("Processed request: %s", reqPath) responseCh <- result } // 응답 대기 (타임아웃 포함) select { case result := <-responseCh: fmt.Fprintln(w, result) case <-time.After(3 * time.Second): http.Error(w, "Request processing timed out", http.StatusGatewayTimeout) } }) // 서버 시작 log.Println("Server starting on :8080") log.Fatal(http.ListenAndServe(":8080", nil)) }