Go 대규모 고루틴 관리 질문

안녕하세요. 질문자님께서 겪고 계신 문제는 Go에서 동시성 프로그래밍을 할 때 흔히 마주치는 중요한 문제입니다. 질문하신 내용에 맞춰 워커 풀(Worker Pool) 패턴과 sync.WaitGroup을 활용한 해결책을 제시해 드리겠습니다.

1. Worker Pool 패턴

워커 풀은 미리 정해진 수의 고루틴(워커)을 만들어 놓고, 이 워커들이 작업 큐(채널)에서 작업을 하나씩 가져와 처리하는 방식입니다. 이를 통해 동시에 실행되는 고루틴의 수를 제어하여 시스템 자원을 안정적으로 사용할 수 있습니다.

핵심 구성 요소:

• jobs 채널: 처리해야 할 작업들을 담아두는 채널입니다.
• results 채널: 워커들이 작업을 완료한 후 결과물을 보내는 채널입니다.
• 워커 고루틴: jobs 채널에서 작업을 꺼내 처리하고 results 채널로 결과를 보냅니다.

아래는 전체적인 구조를 보여주는 예시 코드입니다.

package main

import (
"fmt"
"sync"
"time"
)

// 가상의 작업 데이터
type Job struct {
ID      int
Payload string
}

// 작업 결과
type Result struct {
JobID   int
Message string
Error   error
}

// 워커 함수
func worker(id int, wg *sync.WaitGroup, jobs <-chan Job, results chan<- Result) {
defer wg.Done()
for j := range jobs {
fmt.Printf("워커 %d, 작업 %d 시작\n", id, j.ID)
// 실제 작업 시뮬레이션
time.Sleep(time.Millisecond * 500)
if j.ID%10 == 0 { // 일부러 에러 발생
results <- Result{JobID: j.ID, Message: "에러 발생!", Error: fmt.Errorf("작업 %d 처리 중 에러", j.ID)}
} else {
results <- Result{JobID: j.ID, Message: fmt.Sprintf("작업 %d 완료", j.ID)}
}
fmt.Printf("워커 %d, 작업 %d 종료\n", id, j.ID)
}
}

func main() {
const numJobs = 100
const numWorkers = 5

jobs := make(chan Job, numJobs)
results := make(chan Result, numJobs)

var wg sync.WaitGroup

// 1. 워커 고루틴 생성
for w := 1; w <= numWorkers; w++ {
wg.Add(1)
go worker(w, &wg, jobs, results)
}

// 2. 작업 채널에 작업 할당
for j := 1; j <= numJobs; j++ {
jobs <- Job{ID: j, Payload: "some-data"}
}
close(jobs) // 더 이상 보낼 작업이 없음을 알림

// 3. 모든 워커가 끝날 때까지 대기
wg.Wait()
close(results)

// 4. 결과 수집
fmt.Println("\n--- 결과 ---")
for r := range results {
if r.Error != nil {
fmt.Printf("[실패] Job %d: %s\n", r.JobID, r.Error.Error())
} else {
fmt.Printf("[성공] Job %d: %s\n", r.JobID, r.Message)
}
}
}

2. 질문에 대한 답변 정리

• 고루틴 수 제한: numWorkers 상수를 통해 동시에 실행될 워커의 수를 5개로 제한했습니다. 시스템 사양에 맞게 이 값을 조절하면 됩니다.

• 에러 핸들링: Result 구조체에 Error 필드를 추가하여 작업의 성공 여부와 에러 정보를 함께 전달합니다. 메인 함수에서 results 채널을 순회하며 에러를 중앙에서 처리할 수 있습니다.

• 작업 완료 보장: sync.WaitGroup을 사용합니다. 워커를 생성할 때 wg.Add(1)로 카운터를 늘리고, 각 워커 함수가 종료되기 직전 defer wg.Done()을 호출하여 카운터를 줄입니다. 메인 스레드에서는 wg.Wait()을 통해 모든 워커가 종료될 때까지 기다립니다.

이 패턴을 사용하면 안정적이고 확장 가능한 동시성 파이프라인을 구축할 수 있습니다. 도움이 되셨기를 바랍니다.

좋은 질문입니다. 워커 풀 패턴은 매우 강력하고 표준적인 방법이지만, 때로는 직접 구현하기에 다소 장황할 수 있습니다. 질문자님께서 언급하신 errgroup은 이러한 종류의 작업을 훨씬 간결하고 안전하게 처리할 수 있도록 도와주는 훌륭한 라이브러리입니다.

golang.org/x/sync/errgroup 패키지는 다음과 같은 기능을 제공하여 질문자님의 고민을 해결해줍니다.

• 첫 번째 에러 전파: 그룹 내의 고루틴 중 하나라도 nil이 아닌 에러를 반환하면, 그 에러가 즉시 그룹 전체에 전파됩니다.
• Context를 이용한 취소 전파: WithContext를 사용하면, 한 고루틴에서 에러가 발생했을 때 다른 모든 고루틴에게 취소 신호를 보낼 수 있습니다. 이는 불필요한 작업을 즉시 중단시켜 리소스를 절약하는 데 매우 유용합니다.
• 자동 WaitGroup 관리: 내부적으로 sync.WaitGroup과 유사한 메커니즘을 사용하여 모든 고루틴이 종료될 때까지 기다리는 것을 보장합니다.

errgroup 사용 예시

질문자님의 시나리오에 errgroup을 적용하면 코드가 다음과 같이 바뀝니다.

package main

import (
"context"
"fmt"
"log"

"golang.org/x/sync/errgroup"
)

// 가상의 데이터
type DataItem struct {
ID int
}

func process(item DataItem) (string, error) {
// 일부러 에러 발생
if item.ID == 5 {
return "", fmt.Errorf("치명적인 에러 발생! (ID: %d)", item.ID)
}
return fmt.Sprintf("데이터 %d 처리 성공", item.ID), nil
}

func processDataStreamWithErrGroup(items []DataItem) error {
g, ctx := errgroup.WithContext(context.Background())

for _, item := range items {
// item을 로컬 변수로 복사하여 클로저 문제 방지
currentItem := item

g.Go(func() error {
// 다른 고루틴에서 에러가 발생하여 컨텍스트가 취소되었는지 확인
if ctx.Err() != nil {
log.Printf("작업 취소됨 (ID: %d)", currentItem.ID)
return ctx.Err()
}

result, err := process(currentItem)
if err != nil {
// 에러가 발생하면 그룹에 전파됨
return err
}
log.Println(result)
return nil
})
}

// 모든 g.Go()로 실행된 함수가 끝날 때까지 기다림.
// 만약 어느 한 곳에서라도 에러가 반환되었다면, 그 첫 번째 에러가 반환됨.
if err := g.Wait(); err != nil {
log.Printf("하나 이상의 고루틴에서 에러가 발생했습니다: %v", err)
return err
}

log.Println("모든 작업이 성공적으로 완료되었습니다.")
return nil
}

func main() {
items := make([]DataItem, 10)
for i := 0; i < 10; i++ {
items[i] = DataItem{ID: i + 1}
}

err := processDataStreamWithErrGroup(items)
if err != nil {
fmt.Println("\n최종 결과: 실패")
} else {
fmt.Println("\n최종 결과: 성공")
}
}

고루틴 수 제한하기 (g.SetLimit)

만약 동시에 실행되는 고루틴의 수를 제한하고 싶다면 errgroup의 SetLimit 메서드를 사용하면 됩니다. 이렇게 하면 워커 풀을 직접 구현하는 것과 동일한 효과를 매우 간단하게 얻을 수 있습니다.

func processDataStreamWithErrGroupAndLimit(items []DataItem) error {
    g, ctx := errgroup.WithContext(context.Background())
    // 동시에 최대 10개의 고루틴만 실행하도록 제한
    g.SetLimit(10)

    // ... (나머지 코드는 위와 동일)
}

결론적으로, 복잡한 동기화 로직을 직접 구현하기보다는 errgroup과 같은 검증된 라이브러리를 활용하는 것이 더 안전하고, 간결하며, 유지보수하기 좋은 코드를 작성하는 길입니다. 강력히 추천합니다.