Go의 Worker Pool: Goroutine 제한, 결과 수집, 안전한 취소

1. 개요

Go의 goroutine은 생성 비용이 낮아서, 각 작업마다 go를 호출하는 것이 흔합니다. 작업 수가 수만 개에 달하고 각 작업이 데이터베이스나 HTTP를 호출할 때, 이 방식은 connection pool을 고갈시키거나 외부 서비스의 rate limit을 초과하기 쉽습니다.

이 기사는 worker pool을 구축하는 방법을 안내합니다. 즉, 고정된 수의 goroutine이 채널에서 작업을 받는 방식입니다. 읽고 나면 기본 pool을 작성하고, 결과와 오류를 수집하고, context로 안전하게 취소하고, errgroup.SetLimit으로 단순화하고, pool이 실제로 동시성을 제한하는지 테스트하는 방법을 알게 될 것입니다.

2. 문제: 제한 없는 goroutine

주문 목록을 처리하는 코드를 생각해봅시다:

for _, o := range orders {
	go chargeOrder(o) // 50,000개 주문 -> 50,000개 goroutine이 동시에 payment API 호출
}

각 goroutine은 단지 몇 KB의 메모리만 사용하므로 문제는 보통 RAM에 있지 않습니다. 문제는 백엔드 리소스에 있습니다:

데이터베이스의 connection pool은 보통 20–50개 연결만 가집니다.
Payment gateway는 초당 요청 수를 제한합니다.
프로세스의 파일 디스크립터 수는 제한이 있습니다.

Worker pool은 두 개념을 분리하여 이를 해결합니다. 작업 수는 매우 클 수 있지만, 동시 실행되는 작업 수는 worker 수로 제한됩니다.

3. 기본 Worker Pool

이 모델은 세 부분으로 구성됩니다:

처리해야 할 작업을 포함하는 jobs 채널.
N개의 goroutine worker이며, 각 worker는 채널에서 for range로 반복합니다.
모든 worker가 완료될 때까지 대기하는 sync.WaitGroup.

package main

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

type Order struct {
	ID     int
	Amount float64
}

func worker(id int, orders <-chan Order, wg *sync.WaitGroup) {
	defer wg.Done()
	for o := range orders {
		time.Sleep(100 * time.Millisecond) // payment API 호출 시뮬레이션
		fmt.Printf("worker %d order %d (%.0f) 처리\n", id, o.ID, o.Amount)
	}
}

func main() {
	orders := make(chan Order)
	var wg sync.WaitGroup

	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go worker(i, orders, &wg)
	}

	start := time.Now()
	for i := 1; i <= 9; i++ {
		orders <- Order{ID: i, Amount: float64(i) * 10}
	}
	close(orders)
	wg.Wait()
	fmt.Println("완료 시간", time.Since(start).Round(100*time.Millisecond))
}

코드가 다음과 같이 작동합니다:

orders <-chan Order는 worker 쪽에서 수신만 가능한 채널입니다. 컴파일러는 worker가 실수로 채널로 보내거나 닫으려 하면 컴파일을 거부합니다.
for o := range orders는 채널이 close되고 모든 요소가 소진되면 자동으로 종료됩니다.
close(orders)는 “작업 끝” 신호입니다. 이 줄을 빠뜨리면 worker는 계속 대기하고 wg.Wait()에서 데드락이 발생합니다.
wg.Add(1)은 반드시 go worker(...) 전에 호출해야 하며, goroutine 내부에서 호출하면 안 됩니다.

실행 결과 (줄의 순서는 다를 수 있음):

worker 3 order 1 (10) 처리
worker 1 order 2 (20) 처리
worker 2 order 3 (30) 처리
worker 2 order 4 (40) 처리
...
완료 시간 300ms

9개 주문을 3개 worker로 나누고, 각 주문은 100ms가 걸리므로 총 시간은 순차 실행의 900ms 대신 약 300ms입니다.

참고: Go 1.25부터 wg.Go(func() { worker(i, orders) })로 쓸 수 있습니다. 이 메서드는 자동으로 Add와 Done을 호출하므로 둘 중 하나를 빠뜨리는 실수를 방지합니다.

4. 결과와 오류 수집

실제로는 어떤 주문이 성공하고 어떤 주문이 실패했는지 알아야 합니다. 이를 위해 results 채널을 추가합니다:

package main

import (
	"errors"
	"fmt"
	"sync"
)

type Order struct {
	ID     int
	Amount float64
}

type Result struct {
	OrderID int
	Fee     float64
	Err     error
}

func chargeOrder(o Order) (float64, error) {
	if o.Amount <= 0 {
		return 0, errors.New("유효하지 않은 금액")
	}
	return o.Amount * 0.02, nil // 수수료 2%
}

func main() {
	orders := []Order{{1, 100}, {2, 0}, {3, 250}, {4, 80}}

	jobs := make(chan Order)
	results := make(chan Result)
	var wg sync.WaitGroup

	for range 3 {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for o := range jobs {
				fee, err := chargeOrder(o)
				results <- Result{OrderID: o.ID, Fee: fee, Err: err}
			}
		}()
	}

	go func() {
		for _, o := range orders {
			jobs <- o
		}
		close(jobs)
	}()

	go func() {
		wg.Wait()
		close(results)
	}()

	for r := range results {
		if r.Err != nil {
			fmt.Printf("order %d 오류: %v\n", r.OrderID, r.Err)
			continue
		}
		fmt.Printf("order %d 수수료 %.2f\n", r.OrderID, r.Fee)
	}
}

출력 예:

order 1 수수료 2.00
order 3 수수료 5.00
order 2 오류: 유효하지 않은 금액
order 4 수수료 1.60

주의할 점 세 가지:

작업 전송은 별도 goroutine에서 실행됩니다. main이 같은 goroutine에서 작업을 보내면서 동시에 결과를 읽으려 하면, worker는 results <-에서 블로킹되고 main은 jobs <-에서 블로킹됩니다. 결과는 데드락입니다.
세 번째 goroutine은 wg.Wait()를 기다린 후 close(results)를 합니다. 모든 worker가 완료된 후에만 더 이상 results로 보낼 것이 없음을 확신할 수 있습니다. 일반 규칙: 송신 측만이 채널을 닫습니다.
오류는 Result에 포함되며 전체 pool을 중단시키지 않습니다. 한 주문의 오류가 다른 주문을 막아서는 안 됩니다.

결과의 순서는 입력 순서와 일치하지 않습니다. 순서를 유지해야 하면 작업에 인덱스를 포함시키거나 섹션 6의 방법을 사용하세요.

5. Context로 안전하게 취소

HTTP 요청이 취소되거나 시간 초과되면, pool은 payment API를 계속 호출하는 대신 중지해야 합니다. 이는 worker가 ctx.Done()을 수신 대기하도록 하여 수행합니다:

func worker(ctx context.Context, jobs <-chan Order, results chan<- Result) {
	for {
		select {
		case <-ctx.Done():
			return
		case o, ok := <-jobs:
			if !ok {
				return
			}
			fee, err := chargeOrder(o)
			select {
			case results <- Result{OrderID: o.ID, Fee: fee, Err: err}:
			case <-ctx.Done():
				return
			}
		}
	}
}

producer 측도 마찬가지로 select가 필요합니다:

go func() {
	defer close(jobs)
	for _, o := range orders {
		select {
		case jobs <- o:
		case <-ctx.Done():
			return
		}
	}
}()

그리고 호출 지점에서 timeout이 있는 context를 생성합니다:

ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()

양쪽 송신에서 select가 필요한 이유: worker가 작업을 수신할 때만 ctx를 확인하면, 읽는 쪽이 이미 떠났을 때 results <-에서 계속 블로킹될 수 있습니다. 이것은 매우 흔한 goroutine 누수(goroutine leak)의 원인입니다.

참고: select는 worker가 작업 사이에 빠져나가는 것을 도울 뿐입니다. chargeOrder가 HTTP를 호출하면, 실행 중인 요청까지 취소하려면 ctx를 전달해야 합니다 (http.NewRequestWithContext).

6. errgroup.SetLimit으로 단순화

채널, WaitGroup, context를 직접 작성하기는 실수하기 쉽습니다. errgroup 패키지에는 SetLimit이 있어서, 몇 줄만으로 제한된 pool을 만들 수 있습니다. 먼저 의존성을 추가합니다:

go get golang.org/x/sync/errgroup

아래 generic 함수는 slice를 처리하되 최대 limit goroutine을 사용하고, 결과 순서를 유지하며, 첫 번째 오류에서 중단합니다:

package pool

import (
	"context"

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

func Process[T, R any](
	ctx context.Context,
	limit int,
	items []T,
	fn func(context.Context, T) (R, error),
) ([]R, error) {
	results := make([]R, len(items))
	g, ctx := errgroup.WithContext(ctx)
	g.SetLimit(limit)

	for i, item := range items {
		g.Go(func() error {
			r, err := fn(ctx, item)
			if err != nil {
				return err
			}
			results[i] = r
			return nil
		})
	}

	if err := g.Wait(); err != nil {
		return nil, err
	}
	return results, nil
}

코드가 다음과 같이 작동합니다:

g.SetLimit(limit)은 이미 limit goroutine이 실행 중일 때 g.Go를 블로킹합니다. 효과적으로 worker pool과 동등합니다.
errgroup.WithContext는 함수가 오류를 반환하는 즉시 취소되는 ctx를 반환합니다. ctx를 사용하는 다른 작업들은 중단 신호를 받습니다.
results[i] = r은 각 goroutine이 고유한 인덱스로 쓰므로 동시 실행에서 안전합니다. 이 경우 mutex가 필요하지 않습니다.
Go 1.22부터 루프 변수 i와 item은 각 반복마다 새로 생성되므로, 클로저가 이전 버전처럼 변수를 공유하지 않습니다.

호출 방법:

fees, err := pool.Process(ctx, 5, orders, func(ctx context.Context, o Order) (float64, error) {
	return paymentService.Charge(ctx, o)
})

직접 작성한 pool과의 차이: errgroup은 N개 worker를 재사용하는 대신 각 item마다 새로운 goroutine을 만듭니다. 대부분의 애플리케이션에서는 이 차이가 무시할 수 있습니다. worker가 상태를 유지해야 한다면, 예를 들어 각 worker가 연결을 가지고 있다면, 섹션 3–5의 직접 작성 pool을 사용하세요.

7. Worker Pool 테스트

두 가지 주요 동작을 테스트해야 합니다. 첫째, pool은 절대 limit을 초과하지 않습니다. 둘째, 오류가 올바르게 반환됩니다. 동시성 수준을 측정하기 위해 atomic.Int32를 사용합니다:

package pool

import (
	"context"
	"errors"
	"sync/atomic"
	"testing"
	"time"
)

func TestProcess_LimitsConcurrencyAndKeepsOrder(t *testing.T) {
	var running, maxSeen atomic.Int32
	items := []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}

	got, err := Process(context.Background(), 3, items, func(_ context.Context, n int) (int, error) {
		cur := running.Add(1)
		defer running.Add(-1)
		for {
			m := maxSeen.Load()
			if cur <= m || maxSeen.CompareAndSwap(m, cur) {
				break
			}
		}
		time.Sleep(10 * time.Millisecond)
		return n * 2, nil
	})

	if err != nil {
		t.Fatalf("예상치 못한 오류: %v", err)
	}
	if maxSeen.Load() > 3 {
		t.Errorf("동시 실행 %d, 제한은 3", maxSeen.Load())
	}
	want := []int{2, 4, 6, 8, 10, 12, 14, 16, 18, 20}
	for i := range want {
		if got[i] != want[i] {
			t.Fatalf("got[%d] = %d, want %d", i, got[i], want[i])
		}
	}
}

func TestProcess_ReturnsFirstError(t *testing.T) {
	errDeclined := errors.New("카드가 거부됨")

	_, err := Process(context.Background(), 2, []int{1, 2, 3}, func(_ context.Context, n int) (int, error) {
		if n == 2 {
			return 0, errDeclined
		}
		return n, nil
	})

	if !errors.Is(err, errDeclined) {
		t.Fatalf("errDeclined를 원함, %v를 받음", err)
	}
}

race detector와 함께 테스트를 실행합니다:

go test -race ./...

ok  	example.com/shop/pool	0.061s

CompareAndSwap 루프는 잠금 없이 maxSeen을 업데이트합니다. -race 플래그는 동시 코드에서 필수입니다. 일반적인 눈으로 놓치기 쉬운 공유 데이터 액세스를 감지하기 때문입니다.

8. 흔한 오류와 사용하지 말아야 할 때

흔한 오류들:

close(jobs) 잊음: worker는 절대 종료되지 않고 프로그램은 fatal error: all goroutines are asleep - deadlock!을 보고합니다.
수신 측에서 채널 닫기 또는 두 번 닫기: panic: close of closed channel을 발생시킵니다.
goroutine 내부에서 wg.Add 호출: Wait는 goroutine이 Add할 시간이 없을 때 반환할 수 있습니다.
같은 goroutine에서 작업 전송 및 결과 읽기 (버퍼 없는 채널 사용): 섹션 4에서 분석한 대로 데드락을 발생시킵니다.
결과 전송 시 ctx.Done()에 대해 select 없음: 읽는 쪽이 일찍 종료되었을 때 goroutine 누수를 야기합니다.
worker 수를 임의로 선택: CPU-bound 작업의 경우 runtime.NumCPU()를 사용하세요. I/O-bound 작업의 경우 백엔드 리소스 제한에 따라, 예를 들어 connection pool 크기에 따라 결정하세요.

worker pool을 사용하지 않아야 할 때:

작업 수가 적고 미리 알려져 있는 경우, 예를 들어 3개 서비스를 병렬로 호출할 때. 이 경우 errgroup 없이 SetLimit이면 충분합니다.
극도로 짧은 작업, 단 몇 마이크로초. 채널을 통한 비용이 작업 자체보다 클 수 있으므로, 순차 처리 또는 배치(batch) 처리가 더 효율적입니다.
동시성 수가 아닌 속도(요청/초)로 제한해야 하는 경우. 이 경우 rate limiter가 필요합니다(예: golang.org/x/time/rate). Worker pool은 rate limiter를 대체할 수 없습니다.

9. 결론

Worker pool은 Go에서 동시성 수준을 제어하는 데 도움이 됩니다: 작업 수는 매우 클 수 있지만 백엔드 리소스는 여전히 보호됩니다. 이 기사는 다음 단계를 다루었습니다:

channel과 sync.WaitGroup을 사용한 기본 pool.
results 채널을 통해 결과와 오류를 수집하고, 올바른 시점에 채널을 닫기.
양쪽 송신에서 context로 안전하게 취소.
generic Process 함수를 사용하여 errgroup.SetLimit으로 단순화.
go test -race로 동시성 제한과 오류 테스트.

대부분의 경우 errgroup.SetLimit이 가장 간단하고 오류가 적은 선택입니다. worker가 상태를 유지해야 하거나 작업 및 결과 흐름에 대한 세밀한 제어가 필요할 때만 직접 pool을 작성해야 합니다.

Go의 Worker Pool: Goroutine 제한, 결과 수집, 안전한 취소

1. 개요

2. 문제: 제한 없는 goroutine

3. 기본 Worker Pool

4. 결과와 오류 수집

5. Context로 안전하게 취소

6. errgroup.SetLimit으로 단순화

7. Worker Pool 테스트

8. 흔한 오류와 사용하지 말아야 할 때

9. 결론

같은 주제의 글

Go HTTP 서버의 Graceful Shutdown

Go에서 encoding/json으로 JSON 읽고 쓰기

Go의 struct embedding으로 상속 대체하기

Go에서 defer, panic, recover: 언제 사용할까

Go를 pprof로 프로파일링하기: CPU, 메모리, goroutine 병목 찾기