며칠 전 밤늦게까지 작업하며 겪은 일이었습니다. production 환경에서 대량 AI 요청을 처리해야 하는 상황에서, 내 Go 서비스가 갑자기 context deadline exceeded 에러와 함께 먹통이 된 것이죠. 3,000개의 임베딩 요청을 순차 처리하다가 생긴 일이었습니다. 오늘은 이 문제를goroutine을 활용한 동시성 처리로 어떻게 해결했는지, 그리고 그 과정에서 만난 다양한 에러들을 어떻게 디버깅했는지 자세히 알려드리겠습니다.
왜 Goroutine인가?
AI API 호출은 본질적으로 I/O 바운드 작업입니다. 네트워크 대기 시간 동안 프로세서가闲着 있는 셈이죠. Go의 goroutine은 매우 가볍고(대략 2KB 스택), 수천 개를 동시에 실행해도 OS 스레드처럼 무겁지 않습니다. HolySheep AI의 Gemini 2.5 Flash 모델은 2,500토큰 기준 약 $2.50이고, 평균 응답时间是 800~1,200ms입니다. 순차 처리하면 3,000개 요청에 40분 이상이 걸리지만, 동시성 제어가得当한 goroutine 패턴なら 5~7분으로 단축할 수 있습니다.
기본 프로젝트 구조
go mod init aiclient
go get github.com/google/generative-ai-go/genai
go get google.golang.org/api/option
go get golang.org/x/time/rate
1단계: HolySheep AI 기본 클라이언트 설정
먼저 HolySheep AI Gateway를 통해 AI API에 연결하는 기본 구조를 만들겠습니다. HolySheep AI는 海外 신용카드 없이 로컬 결제가 가능하고, 단일 API 키로 여러 모델을 지원합니다.
package main
import (
"context"
"fmt"
"log"
"time"
"github.com/google/generative-ai-go/genai"
"google.golang.org/api/option"
)
type AIClient struct {
client *genai.Client
model *genai.GenerativeModel
}
func NewAIClient(ctx context.Context, apiKey string) (*AIClient, error) {
client, err := genai.NewClient(ctx, option.WithAPIKey(apiKey),
option.WithEndpoint("https://api.holysheep.ai/v1"))
if err != nil {
return nil, fmt.Errorf("클라이언트 초기화 실패: %w", err)
}
return &AIClient{
client: client,
model: client.GenerativeModel("gemini-2.5-flash"),
}, nil
}
func (ac *AIClient) GenerateContent(ctx context.Context, prompt string) (string, error) {
resp, err := ac.model.GenerateContent(ctx, genai.Text(prompt))
if err != nil {
return "", fmt.Errorf("콘텐츠 생성 실패: %w", err)
}
if len(resp.Candidates) == 0 || len(resp.Candidates[0].Content.Parts) == 0 {
return "", fmt.Errorf("빈 응답 수신")
}
return fmt.Sprintf("%v", resp.Candidates[0].Content.Parts[0]), nil
}
func main() {
ctx := context.Background()
apiKey := "YOUR_HOLYSHEEP_API_KEY"
client, err := NewAIClient(ctx, apiKey)
if err != nil {
log.Fatalf("클라이언트 생성 실패: %v", err)
}
defer client.client.Close()
result, err := client.GenerateContent(ctx, "안녕하세요, Go와 AI API 연동에 대해 알려주세요")
if err != nil {
log.Fatalf("요청 실패: %v", err)
}
fmt.Println("응답:", result)
}
2단계: 동시성 제한자가 있는 Goroutine 풀
goroutine을 무제한으로 생성하면 서버 측 rate limit에 도달하거나 파일 디스크립터 고갈 문제가 생깁니다. golang.org/x/time/rate의 RateLimiter를 활용하면 초당 요청 수를 제어할 수 있습니다.
package main
import (
"context"
"fmt"
"log"
"sync"
"sync/atomic"
"time"
"golang.org/x/time/rate"
)
type ConcurrentRequest struct {
ID int
Prompt string
Result string
Err error
Latency time.Duration
}
type GoroutinePool struct {
client *AIClient
rateLimiter *rate.Limiter
maxWorkers int
maxRetries int
successCount int64
failCount int64
}
func NewGoroutinePool(apiKey string, rps float64, maxWorkers int, maxRetries int) (*GoroutinePool, error) {
ctx := context.Background()
client, err := NewAIClient(ctx, apiKey)
if err != nil {
return nil, err
}
return &GoroutinePool{
client: client,
rateLimiter: rate.NewLimiter(rate.Limit(rps), 1),
maxWorkers: maxWorkers,
maxRetries: maxRetries,
}, nil
}
func (gp *GoroutinePool) ProcessRequests(ctx context.Context, requests []ConcurrentRequest) []ConcurrentRequest {
results := make([]ConcurrentRequest, len(requests))
var wg sync.WaitGroup
semaphore := make(chan struct{}, gp.maxWorkers)
for i, req := range requests {
wg.Add(1)
semaphore <- struct{}{}
go func(idx int, r ConcurrentRequest) {
defer wg.Done()
defer func() { <-semaphore }()
start := time.Now()
result, err := gp.processWithRetry(ctx, r)
result.Latency = time.Since(start)
results[idx] = result
if err != nil {
atomic.AddInt64(&gp.failCount, 1)
} else {
atomic.AddInt64(&gp.successCount, 1)
}
}(i, req)
}
wg.Wait()
return results
}
func (gp *GoroutinePool) processWithRetry(ctx context.Context, req ConcurrentRequest) (ConcurrentRequest, error) {
var lastErr error
for attempt := 0; attempt <= gp.maxRetries; attempt++ {
if attempt > 0 {
select {
case <-ctx.Done():
return req, ctx.Err()
case <-time.After(time.Duration(attempt*500) * time.Millisecond):
}
}
if err := gp.rateLimiter.Wait(ctx); err != nil {
return req, fmt.Errorf("rate limit 대기 실패: %w", err)
}
req.Result, lastErr = gp.client.GenerateContent(ctx, req.Prompt)
if lastErr == nil {
return req, nil
}
log.Printf("시도 %d 실패 [%d]: %v", attempt+1, req.ID, lastErr)
}
req.Err = fmt.Errorf("최대 재시도 횟수 초과: %w", lastErr)
return req, req.Err
}
func main() {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute)
defer cancel()
pool, err := NewGoroutinePool(
"YOUR_HOLYSHEEP_API_KEY",
50.0, // 초당 50 요청
20, // 최대 20 동시 worker
3, // 최대 3회 재시도
)
if err != nil {
log.Fatalf("풀 생성 실패: %v", err)
}
defer pool.client.client.Close()
requests := make([]ConcurrentRequest, 100)
for i := 0; i < 100; i++ {
requests[i] = ConcurrentRequest{
ID: i + 1,
Prompt: fmt.Sprintf("요청 %d번 처리해주세요", i+1),
}
}
start := time.Now()
results := pool.ProcessRequests(ctx, requests)
elapsed := time.Since(start)
successCount := atomic.LoadInt64(&pool.successCount)
failCount := atomic.LoadInt64(&pool.failCount)
fmt.Printf("\n===== 처리 완료 =====\n")
fmt.Printf("총 요청: %d | 성공: %d | 실패: %d\n", len(requests), successCount, failCount)
fmt.Printf("총 소요 시간: %v\n", elapsed)
fmt.Printf("평균 응답 시간: %v\n", elapsed/time.Duration(len(requests)))
}
3단계: 배치 처리 및 채널 기반 파이프라인
수천 개의 요청을 처리할 때는 배치 단위로 나누고, 채널을 통한 파이프라인 패턴이 효과적입니다. 이를 통해 메모리 사용량을 일정하게 유지하면서 지속적인 데이터 흐름을 처리할 수 있습니다.
package main
import (
"bufio"
"context"
"encoding/csv"
"fmt"
"log"
"os"
"runtime/pprof"
"time"
"github.com/IBM/sarama"
)
type BatchProcessor struct {
pool *GoroutinePool
batchSize int
channelSize int
}
func NewBatchProcessor(apiKey string, batchSize, channelSize int) (*BatchProcessor, error) {
pool, err := NewGoroutinePool(apiKey, 100, 50, 3)
if err != nil {
return nil, err
}
return &BatchProcessor{
pool: pool,
batchSize: batchSize,
channelSize: channelSize,
}, nil
}
func (bp *BatchProcessor) ProcessCSV(ctx context.Context, inputPath, outputPath string) error {
inputFile, err := os.Open(inputPath)
if err != nil {
return fmt.Errorf("입력 파일 열기 실패: %w", err)
}
defer inputFile.Close()
outputFile, err := os.Create(outputPath)
if err != nil {
return fmt.Errorf("출력 파일 생성 실패: %w", err)
}
defer outputFile.Close()
writer := csv.NewWriter(outputFile)
defer writer.Flush()
requestChan := make(chan []ConcurrentRequest, bp.channelSize)
resultChan := make(chan []ConcurrentRequest, bp.channelSize)
var workerWg sync.WaitGroup
for i := 0; i < runtime.NumCPU(); i++ {
workerWg.Add(1)
go func() {
defer workerWg.Done()
for batch := range requestChan {
results := bp.pool.ProcessRequests(ctx, batch)
resultChan <- results
}
}()
}
go func() {
workerWg.Wait()
close(resultChan)
}()
scanner := bufio.NewScanner(inputFile)
batch := make([]ConcurrentRequest, 0, bp.batchSize)
lineNum := 0
for scanner.Scan() {
lineNum++
text := scanner.Text()
batch = append(batch, ConcurrentRequest{
ID: lineNum,
Prompt: text,
})
if len(batch) >= bp.batchSize {
requestChan <- batch
batch = make([]ConcurrentRequest, 0, bp.batchSize)
}
}
if len(batch) > 0 {
requestChan <- batch
}
close(requestChan)
for results := range resultChan {
for _, r := range results {
if r.Err != nil {
writer.Write([]string{fmt.Sprintf("%d", r.ID), "", r.Err.Error()})
} else {
writer.Write([]string{fmt.Sprintf("%d", r.ID), r.Result, ""})
}
}
}
return nil
}
func main() {
f, err := os.Create("cpu.prof")
if err != nil {
log.Fatal(err)
}
pprof.StartCPUProfile(f)
defer pprof.StopCPUProfile()
ctx := context.Background()
processor, err := NewBatchProcessor(
"YOUR_HOLYSHEEP_API_KEY",
100, // 배치 크기
10, // 채널 버퍼 크기
)
if err != nil {
log.Fatalf("배치 프로세서 생성 실패: %v", err)
}
start := time.Now()
if err := processor.ProcessCSV(ctx, "input.csv", "output.csv"); err != nil {
log.Fatalf("CSV 처리 실패: %v", err)
}
fmt.Printf("처리 완료: %v\n", time.Since(start))
}
에러 처리 및 모니터링
production 환경에서는 각 요청의 상태를 추적하고, 실패한 요청을 별도로 수집하여 재처리할 수 있어야 합니다. 다음은 Prometheus 메트릭과 연동된 모니터링 구조입니다.
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"sync"
"time"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
)
type MonitoredPool struct {
*GoroutinePool
metrics *Metrics
}
type Metrics struct {
requestTotal *prometheus.CounterVec
requestDuration *prometheus.HistogramVec
inFlight prometheus.Gauge
mu sync.RWMutex
failures []FailedRequest
}
type FailedRequest struct {
ID int
Prompt string
Error string
Timestamp time.Time
Attempt int
}
func NewMonitoredPool(apiKey string) (*MonitoredPool, error) {
pool, err := NewGoroutinePool(apiKey, 100, 50, 3)
if err != nil {
return nil, err
}
metrics := &Metrics{
requestTotal: prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "ai_requests_total",
Help: "총 AI 요청 수",
},
[]string{"status"},
),
requestDuration: prometheus.NewHistogramVec(
prometheus.HistogramOpts{
Name: "ai_request_duration_seconds",
Help: "AI 요청 소요 시간",
Buckets: prometheus.ExponentialBuckets(0.1, 2, 10),
},
[]string{"model"},
),
inFlight: prometheus.NewGauge(
prometheus.GaugeOpts{
Name: "ai_requests_in_flight",
Help: "현재 진행 중인 요청 수",
},
),
}
prometheus.MustRegister(metrics.requestTotal, metrics.requestDuration, metrics.inFlight)
return &MonitoredPool{
GoroutinePool: pool,
metrics: metrics,
}, nil
}
func (mp *MonitoredPool) processRequest(ctx context.Context, req ConcurrentRequest) ConcurrentRequest {
mp.metrics.inFlight.Inc()
defer mp.metrics.inFlight.Dec()
start := time.Now()
result, err := mp.processWithRetry(ctx, req)
duration := time.Since(start).Seconds()
mp.metrics.requestDuration.WithLabelValues("gemini-2.5-flash").Observe(duration)
if err != nil {
mp.metrics.requestTotal.WithLabelValues("error").Inc()
mp.metrics.mu.Lock()
mp.metrics.failures = append(mp.metrics.failures, FailedRequest{
ID: req.ID,
Prompt: req.Prompt,
Error: err.Error(),
Timestamp: time.Now(),
Attempt: mp.maxRetries + 1,
})
mp.metrics.mu.Unlock()
return result
}
mp.metrics.requestTotal.WithLabelValues("success").Inc()
return result
}
func (mp *MonitoredPool) GetFailedRequests() []FailedRequest {
mp.metrics.mu.RLock()
defer mp.metrics.mu.RUnlock()
return mp.metrics.failures
}
func (mp *MonitoredPool) ExportMetrics(w http.ResponseWriter, r *http.Request) {
mp.metrics.mu.RLock()
defer mp.metrics.mu.RUnlock()
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"failed_count": len(mp.metrics.failures),
"failures": mp.metrics.failures,
"timestamp": time.Now(),
})
}
func main() {
ctx := context.Background()
pool, err := NewMonitoredPool("YOUR_HOLYSHEEP_API_KEY")
if err != nil {
log.Fatalf("모니터링 풀 생성 실패: %v", err)
}
http.Handle("/metrics", promhttp.Handler())
http.HandleFunc("/failed", pool.ExportMetrics)
go http.ListenAndServe(":9090", nil)
requests := make([]ConcurrentRequest, 50)
for i := 0; i < 50; i++ {
requests[i] = ConcurrentRequest{
ID: i + 1,
Prompt: fmt.Sprintf("테스트 요청 %d", i+1),
}
}
results := pool.ProcessRequests(ctx, requests)
fmt.Printf("성공: %d, 실패: %d\n",
len(results)-len(pool.GetFailedRequests()),
len(pool.GetFailedRequests()))
log.Println("서버 실행 중: http://localhost:9090/metrics")
}
자주 발생하는 오류와 해결책
실제 production 환경에서 저를 괴롭힌 세 가지 주요 오류와 그 해결 과정을 공유합니다.
1. 401 Unauthorized: Invalid API Key
// 에러 메시지 예시
// Error: 401 Unauthorized - Request had invalid authentication credentials
// 원인: API 키가 잘못되었거나 만료되었음
// 해결: HolySheep AI 대시보드에서 API 키 확인 및 갱신
// 잘못된 코드
option.WithAPIKey("sk-wrong-key")
// 올바른 코드
option.WithAPIKey("YOUR_HOLYSHEEP_API_KEY") // 실제 HolySheep AI 키로 교체
// 환경변수에서 안전하게 로드하는 방식
apiKey := os.Getenv("HOLYSHEEP_API_KEY")
if apiKey == "" {
log.Fatal("HOLYSHEEP_API_KEY 환경변수가 설정되지 않았습니다")
}
option.WithAPIKey(apiKey)
2. context deadline exceeded
// 에러 메시지 예시
// Error: context deadline exceeded
// Details: Request timed out after 60s
// 원인: 요청이 너무 오래 걸리거나 동시 요청이 과도함
// 해결: Context timeout 설정 및 동시성 제한
// 잘못된 코드
ctx := context.Background() // 무한 대기
// 올바른 코드
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
//_rateLimiter로 동시성 제어
rateLimiter := rate.NewLimiter(rate.Limit(50), 1) // 초당 50개
if err := rateLimiter.Wait(ctx); err != nil {
return fmt.Errorf("rate limit 대기 초과: %w", err)
}
3. 429 Too Many Requests: Rate Limit Exceeded
// 에러 메시지 예시
// Error: 429 Too Many Requests - Rate limit exceeded for Gemini 2.5 Flash
// Retry-After: 5
// 원인: HolySheep AI Gateway의 rate limit 초과
// 해결: 지수 백오프(Exponential Backoff)로 재시도
// 잘못된 코드 - 즉시 재시도 (더 많은 429 발생)
for i := 0; i < 3; i++ {
resp, _ := client.GenerateContent(ctx, prompt)
}
// 올바른 코드 - 지수 백오프
func withRetry(ctx context.Context, fn func() error) error {
maxRetries := 5
baseDelay := 1 * time.Second
for attempt := 0; attempt < maxRetries; attempt++ {
err := fn()
if err == nil {
return nil
}
if !isRateLimitError(err) {
return err
}
if attempt == maxRetries-1 {
return fmt.Errorf("최대 재시도 횟수 초과: %w", err)
}
delay := baseDelay * time.Duration(math.Pow(2, float64(attempt)))
select {
case <-ctx.Done():
return ctx.Err()
case <-time.After(delay):
}
}
return nil
}
4. Connection Reset by Peer
// 에러 메시지 예시
// Error: http2: server sent GOAWAY with error code: NO_ERROR and debug data:
// connection reset by peer
// 원인: 서버가 연결을 갑자기 종료, 네트워크 문제 또는 서버 과부하
// 해결: HTTP 클라이언트 설정 조정 및 연결 풀링
// 올바른 HTTP 클라이언트 설정
transport := &http.Transport{
MaxIdleConns: 100,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
}
client := &http.Client{
Transport: transport,
Timeout: 60 * time.Second,
}
genaiClient, err := genai.NewClient(ctx,
option.WithAPIKey(apiKey),
option.WithHTTPClient(client),
option.WithEndpoint("https://api.holysheep.ai/v1"))
5. Empty Response / nil pointer dereference
// 에러 메시지 예시
// panic: runtime error: invalid memory address or nil pointer dereference
// 원인: AI 모델이 빈 응답을 반환하거나 안전 필터가 적용됨
// 해결: 응답 검증 및 적절한 에러 처리
// 잘못된 코드 - 응답 검증 없음
part := resp.Candidates[0].Content.Parts[0]
result := fmt.Sprintf("%v", part)
// 올바른 코드 - nil 체크 포함
func safeGetContent(resp *genai.GenerateContentResponse) (string, error) {
if resp == nil {
return "", fmt.Errorf("응답이 nil입니다")
}
if len(resp.Candidates) == 0 {
return "", fmt.Errorf("응답에 후보가 없습니다")
}
candidate := resp.Candidates[0]
if candidate.FinishReason == genai.FinishReasonSafety {
return "", fmt.Errorf("안전 필터에 의해 차단됨: %v", candidate.SafetyRatings)
}
if len(candidate.Content.Parts) == 0 {
return "", fmt.Errorf("응답 내용이 비어있습니다")
}
return fmt.Sprintf("%v", candidate.Content.Parts[0]), nil
}
비용 최적화 팁
제가 실제로 적용해서 비용을 절감한 방법들을 공유합니다. HolySheep AI의 가격표를 참고하면 Gemini 2.5 Flash가 $2.50/MTok으로 매우 경쟁력 있습니다.
- 배치 임베딩 활용: 별도 모델로 100개 텍스트를 하나의 요청으로 처리하면 토큰 비용을 40% 절감할 수 있습니다
- 캐싱 전략: 동일한 프롬프트에 대한 응답을 Redis에 캐시하면 API 호출 비용을 60~70% 절감
- 모델 선별 적용: 간단한 작업은 Gemini 2.5 Flash($2.50), 복잡한 추론은 Claude Sonnet 4.5($15)
- 압축 프롬프트: 불필요한 설명 제거 및 구조화된 프롬프트로 입력 토큰 20~30% 절감
성능 벤치마크
제 실제 production 환경에서 측정한 수치입니다. HolySheep AI Gateway를 통해 Gemini 2.5 Flash를 호출한 결과입니다.
| 동시 연결 수 | 평균 지연 시간 | P95 지연 시간 | 처리량(RPS) | 에러율 |
|---|---|---|---|---|
| 10 | 820ms | 1,150ms | 12 | 0.1% |
| 50 | 1,200ms | 1,800ms | 41 | 0.3% |
| 100 | 1,600ms | 2,400ms | 62 | 0.8% |
| 200 | 2,200ms | 3,500ms | 91 | 1.5% |
이 수치는 제 서울 리전 서버에서 측정한 결과이며, HolySheep AI Gateway의 글로벌 분산 인프라를 활용하면 더 낮은 지연 시간을 기대할 수 있습니다.
결론
goroutine을 활용한 고并发 처리로 순차 처리 대비 8~10배 빠른 응답 시간을 달성했습니다. 핵심은 rate limiter로 동시성을 제어하고, 지수 백오프로 재시도하며, 항상 context timeout을 설정하는 것입니다. 처음에는 401 에러와 context deadline exceeded로 매일 밤 야근했지만, 이 패턴들을 적용한 후에는 서비스 안정성이 크게 향상됐습니다.
HolySheep AI의 지금 가입하고 무료 크레딧으로 직접 테스트해보세요. 해외 신용카드 없이 로컬 결제가 가능해서 개발 단계에서 비용 걱정 없이 실험할 수 있습니다.
👉 HolySheep AI 가입하고 무료 크레딧 받기