เมื่อวานนี้ผมเจอปัญหาหนักใจกับระบบที่ต้องประมวลผลคำขอ AI จำนวนมาก คำสั่งนี้ปรากฏขึ้นบนหน้าจอ:

2024/01/15 14:32:18 ERROR: ConnectionError: dial tcp: i/o timeout after 30s
2024/01/15 14:32:18 ERROR: ConnectionError: dial tcp: i/o timeout after 30s
2024/01/15 14:32:19 ERROR: 401 Unauthorized - Invalid API key or expired token
2024/01/15 14:32:20 ERROR: ConnectionError: dial tcp: i/o timeout after 30s

ทีมของผมต้องเรียก AI API เพื่อวิเคราะห์ข้อความลูกค้า 10,000 รายการต่อวัน แต่วิธีเดิมที่ใช้ sequential request ใช้เวลานานเกินไป จึงตัดสินใจศึกษาการใช้ goroutine เพื่อเพิ่มประสิทธิภาพ ในบทความนี้จะแชร์ประสบการณ์ตรงพร้อมโค้ดที่ใช้งานได้จริง

ทำไมต้องใช้ HolySheep AI

ก่อนจะเข้าสู่เนื้อหาหลัก ผมอยากแนะนำ สมัครที่นี่ HolySheep AI ซึ่งเป็น API gateway ที่รวม AI หลายตัวไว้ที่เดียว มีจุดเด่นที่สำคัญ:

โครงสร้างพื้นฐาน: HTTP Client พร้อม Timeout

ก่อนจะไปถึงเรื่อง concurrency เราต้องตั้งค่า HTTP client ให้ถูกต้องก่อน:

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "fmt"
    "net/http"
    "time"
)

// Config โครงสร้างการตั้งค่า HolySheep API
type Config struct {
    BaseURL string
    APIKey  string
    Timeout time.Duration
}

// HolySheepClient HTTP client สำหรับเรียก HolySheep API
type HolySheepClient struct {
    config  Config
    client  *http.Client
}

// NewHolySheepClient สร้าง client ใหม่
func NewHolySheepClient(apiKey string) *HolySheepClient {
    return &HolySheepClient{
        config: Config{
            BaseURL: "https://api.holysheep.ai/v1",
            APIKey:  apiKey,
            Timeout: 60 * time.Second, // Timeout ยาวพอสำหรับ AI API
        },
        client: &http.Client{
            Timeout: 60 * time.Second,
            Transport: &http.Transport{
                MaxIdleConns:        100,
                MaxIdleConnsPerHost: 100,
                IdleConnTimeout:     90 * time.Second,
            },
        },
    }
}

// ChatRequest โครงสร้าง request สำหรับ Chat API
type ChatRequest struct {
    Model    string        json:"model"
    Messages []ChatMessage json:"messages"
    MaxTokens int          json:"max_tokens,omitempty"
}

type ChatMessage struct {
    Role    string json:"role"
    Content string json:"content"
}

// ChatResponse โครงสร้าง response จาก Chat API
type ChatResponse struct {
    ID      string   json:"id"
    Model   string   json:"model"
    Choices []Choice json:"choices"
}

type Choice struct {
    Message      ChatMessage json:"message"
    FinishReason string      json:"finish_reason"
}

// ChatCompletion เรียก Chat API
func (c *HolySheepClient) ChatCompletion(ctx context.Context, req ChatRequest) (*ChatResponse, error) {
    url := c.config.BaseURL + "/chat/completions"
    
    jsonData, err := json.Marshal(req)
    if err != nil {
        return nil, fmt.Errorf("failed to marshal request: %w", err)
    }
    
    httpReq, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(jsonData))
    if err != nil {
        return nil, fmt.Errorf("failed to create request: %w", err)
    }
    
    httpReq.Header.Set("Content-Type", "application/json")
    httpReq.Header.Set("Authorization", "Bearer "+c.config.APIKey)
    
    resp, err := c.client.Do(httpReq)
    if err != nil {
        return nil, fmt.Errorf("request failed: %w", err)
    }
    defer resp.Body.Close()
    
    if resp.StatusCode != http.StatusOK {
        return nil, fmt.Errorf("API error: status %d", resp.StatusCode)
    }
    
    var chatResp ChatResponse
    if err := json.NewDecoder(resp.Body).Decode(&chatResp); err != nil {
        return nil, fmt.Errorf("failed to decode response: %w", err)
    }
    
    return &chatResp, nil
}

ระบบ High Concurrency ด้วย Goroutine และ Worker Pool

ต่อไปนี้คือหัวใจหลักของบทความ เราจะสร้างระบบที่ประมวลผลหลาย request พร้อมกันอย่างปลอดภัย:

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "time"
    
    "golang.org/x/time/rate"
)

// Request ข้อมูล request ที่ต้องประมวลผล
type Request struct {
    ID      string
    Prompt  string
    Model   string
}

// Result ผลลัพธ์จากการประมวลผล
type Result struct {
    RequestID    string
    Response     string
    LatencyMs    int64
    Error        error
}

// ConcurrentProcessor ระบบประมวลผลแบบ concurrent
type ConcurrentProcessor struct {
    client      *HolySheepClient
    workerCount int
    rateLimiter *rate.Limiter
    results     chan Result
    wg          sync.WaitGroup
    mu          sync.Mutex
    stats       ProcessorStats
}

// ProcessorStats สถิติการทำงาน
type ProcessorStats struct {
    TotalRequests   int
    SuccessCount    int
    ErrorCount      int
    TotalLatencyMs  int64
}

// NewConcurrentProcessor สร้าง processor ใหม่
func NewConcurrentProcessor(apiKey string, workerCount int, requestsPerSecond float64) *ConcurrentProcessor {
    return &ConcurrentProcessor{
        client:      NewHolySheepClient(apiKey),
        workerCount: workerCount,
        rateLimiter: rate.NewLimiter(rate.Limit(requestsPerSecond), workerCount),
        results:     make(chan Result, workerCount*2),
    }
}

// ProcessBatch ประมวลผล batch ของ requests
func (p *ConcurrentProcessor) ProcessBatch(ctx context.Context, requests []Request) []Result {
    startTime := time.Now()
    
    // สร้าง worker goroutines
    taskQueue := make(chan Request, len(requests))
    
    for i := 0; i < p.workerCount; i++ {
        p.wg.Add(1)
        go p.worker(ctx, taskQueue, i)
    }
    
    // ส่ง requests เข้าคิว
    go func() {
        for _, req := range requests {
            taskQueue <- req
        }
        close(taskQueue)
    }()
    
    // รอให้ workers ทำงานเสร็จ
    p.wg.Wait()
    close(p.results)
    
    // เก็บผลลัพธ์
    var results []Result
    for result := range p.results {
        results = append(results, result)
        
        // อัพเดทสถิติ
        p.mu.Lock()
        p.stats.TotalRequests++
        if result.Error == nil {
            p.stats.SuccessCount++
        } else {
            p.stats.ErrorCount++
            log.Printf("Request %s failed: %v", result.RequestID, result.Error)
        }
        p.stats.TotalLatencyMs += result.LatencyMs
        p.mu.Unlock()
    }
    
    elapsed := time.Since(startTime)
    p.printStats(elapsed)
    
    return results
}

// worker โครงสร้าง worker ที่ประมวลผล request
func (p *ConcurrentProcessor) worker(ctx context.Context, tasks <-chan Request, workerID int) {
    defer p.wg.Done()
    
    for req := range tasks {
        // รอ until rate limit allows
        if err := p.rateLimiter.Wait(ctx); err != nil {
            p.results <- Result{
                RequestID: req.ID,
                Error:     fmt.Errorf("rate limiter context: %w", err),
            }
            continue
        }
        
        // วัดเวลา latency
        startTime := time.Now()
        
        // เรียก API
        chatReq := ChatRequest{
            Model: req.Model,
            Messages: []ChatMessage{
                {Role: "user", Content: req.Prompt},
            },
            MaxTokens: 1000,
        }
        
        resp, err := p.client.ChatCompletion(ctx, chatReq)
        latencyMs := time.Since(startTime).Milliseconds()
        
        // ส่งผลลัพธ์กลับ
        result := Result{
            RequestID: req.ID,
            LatencyMs: latencyMs,
        }
        
        if err != nil {
            result.Error = err
        } else if len(resp.Choices) > 0 {
            result.Response = resp.Choices[0].Message.Content
        }
        
        p.results <- result
    }
}

// printStats แสดงสถิติการทำงาน
func (p *ConcurrentProcessor) printStats(elapsed time.Duration) {
    p.mu.Lock()
    defer p.mu.Unlock()
    
    avgLatency := float64(0)
    if p.stats.SuccessCount > 0 {
        avgLatency = float64(p.stats.TotalLatencyMs) / float64(p.stats.SuccessCount)
    }
    
    successRate := float64(0)
    if p.stats.TotalRequests > 0 {
        successRate = float64(p.stats.SuccessCount) / float64(p.stats.TotalRequests) * 100
    }
    
    throughput := float64(p.stats.TotalRequests) / elapsed.Seconds()
    
    fmt.Printf("\n========== สถิติการประมวลผล ==========\n")
    fmt.Printf("รวม Requests:        %d\n", p.stats.TotalRequests)
    fmt.Printf("สำเร็จ:              %d (%.1f%%)\n", p.stats.SuccessCount, successRate)
    fmt.Printf("ล้มเหลว:             %d\n", p.stats.ErrorCount)
    fmt.Printf("Latency เฉลี่ย:      %.1f ms\n", avgLatency)
    fmt.Printf("Throughput:          %.1f req/s\n", throughput)
    fmt.Printf("เวลารวม:             %.1f วินาที\n", elapsed.Seconds())
    fmt.Printf("=====================================\n")
}

ตัวอย่างการใช้งานจริง: วิเคราะห์ข้อความลูกค้า 10,000 รายการ

package main

import (
    "context"
    "encoding/csv"
    "fmt"
    "log"
    "os"
    "time"
)

func main() {
    // ตั้งค่า API Key จาก Environment Variable
    apiKey := os.Getenv("HOLYSHEEP_API_KEY")
    if apiKey == "" {
        apiKey = "YOUR_HOLYSHEEP_API_KEY" // แทนที่ด้วย key จริง
    }
    
    // สร้าง processor ที่มี 50 workers และ rate limit 100 req/s
    processor := NewConcurrentProcessor(apiKey, 50, 100)
    
    // เตรียม requests
    var requests []Request
    
    // อ่านข้อมูลจาก CSV file
    file, err := os.Open("customer_messages.csv")
    if err != nil {
        log.Fatalf("ไม่สามารถเปิดไฟล์: %v", err)
    }
    defer file.Close()
    
    reader := csv.NewReader(file)
    records, err := reader.ReadAll()
    if err != nil {
        log.Fatalf("ไม่สามารถอ่าน CSV: %v", err)
    }
    
    for i, record := range records[1:] { // ข้าม header
        if len(record) < 2 {
            continue
        }
        
        requests = append(requests, Request{
            ID:      fmt.Sprintf("req_%05d", i+1),
            Prompt:  fmt.Sprintf("วิเคราะห์ความรู้สึกของข้อความนี้และจัดหมวดหมู่: %s", record[1]),
            Model:   "gpt-4.1", // ใช้ model ที่ต้องการ
        })
    }
    
    fmt.Printf("เริ่มประมวลผล %d requests...\n", len(requests))
    startTime := time.Now()
    
    // สร้าง context พร้อม timeout
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
    defer cancel()
    
    // ประมวลผล
    results := processor.ProcessBatch(ctx, requests)
    
    // บันทึกผลลัพธ์
    outputFile, err := os.Create("results.csv")
    if err != nil {
        log.Fatalf("ไม่สามารถสร้างไฟล์ผลลัพธ์: %v", err)
    }
    defer outputFile.Close()
    
    writer := csv.NewWriter(outputFile)
    writer.Write([]string{"RequestID", "Response", "LatencyMs", "Status"})
    
    for _, r := range results {
        status := "SUCCESS"
        if r.Error != nil {
            status = fmt.Sprintf("ERROR: %v", r.Error)
        }
        writer.Write([]string{r.RequestID, r.Response, fmt.Sprintf("%d", r.LatencyMs), status})
    }
    writer.Flush()
    
    fmt.Printf("\nเสร็จสิ้น! ใช้เวลาทั้งหมด: %.1f วินาที\n", time.Since(startTime).Seconds())
}

การ Implement Retry Logic อัตโนมัติ

ในการใช้งานจริง บางครั้ง request อาจล้มเหลวชั่วคราว ผมแนะนำให้เพิ่ม retry logic:

// RetryConfig การตั้งค่า retry
type RetryConfig struct {
    MaxAttempts int
    InitialDelay time.Duration
    MaxDelay     time.Duration
    Multiplier   float64
}

// RetryWithBackoff ลอง request ซ้ำเมื่อล้มเหลว
func RetryWithBackoff(ctx context.Context, config RetryConfig, fn func() error) error {
    var err error
    delay := config.InitialDelay
    
    for attempt := 0; attempt < config.MaxAttempts; attempt++ {
        if attempt > 0 {
            fmt.Printf("ลองใหม่ครั้งที่ %d หลังจาก %v...\n", attempt, delay)
            
            select {
            case <-time.After(delay):
                // รอตามเวลาที่กำหนด
            case <-ctx.Done():
                return ctx.Err()
            }
            
            // เพิ่ม delay แบบ exponential backoff
            delay = time.Duration(float64(delay) * config.Multiplier)
            if delay > config.MaxDelay {
                delay = config.MaxDelay
            }
        }
        
        err = fn()
        if err == nil {
            return nil
        }
        
        // ตรวจสอบว่า error นี้ควร retry หรือไม่
        if !isRetryableError(err) {
            return err
        }
    }
    
    return fmt.Errorf("retry exhausted after %d attempts: %w", config.MaxAttempts, err)
}

// isRetryableError ตรวจสอบว่า error ควร retry หรือไม่
func isRetryableError(err error) bool {
    if err == nil {
        return false
    }
    
    errStr := err.Error()
    retryableErrors := []string{
        "timeout",
        "i/o timeout",
        "connection refused",
        "connection reset",
        "network",
        "503",
        "502",
        "429", // Too Many Requests
    }
    
    for _, pattern := range retryableErrors {
        if contains(errStr, pattern) {
            return true
        }
    }
    return false
}

func contains(s, substr string) bool {
    return len(s) >= len(substr) && (s == substr || len(s) > 0 && containsHelper(s, substr))
}

func containsHelper(s, substr string) bool {
    for i := 0; i <= len(s)-len(substr); i++ {
        if s[i:i+len(substr)] == substr {
            return true
        }
    }
    return false
}

ข้อผิดพลาดที่พบบ่อยและวิธีแก้ไข

1. 401 Unauthorized - Invalid API key or expired token

สาเหตุ: API key ไม่ถูกต้องหรือหมดอายุ ซึ่งเป็นปัญหาที่พบบ่อยที่สุด

วิธีแก้ไข:

// ตรวจสอบ API Key ก่อนเริ่มทำงาน
func validateAPIKey(apiKey string) error {
    if apiKey == "" {
        return fmt.Errorf("API key is empty - please set HOLYSHEEP_API_KEY environment variable")
    }
    
    if apiKey == "YOUR_HOLYSHEEP_API_KEY" {
        return fmt.Errorf("please replace YOUR_HOLYSHEEP_API_KEY with your actual API key from https://www.holysheep.ai/register")
    }
    
    // ทดสอบ API key ด้วย simple request
    client := NewHolySheepClient(apiKey)
    ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
    defer cancel()
    
    req := ChatRequest{
        Model: "gpt-4.1",
        Messages: []ChatMessage{
            {Role: "user", Content: "Hi"},
        },
        MaxTokens: 5,
    }
    
    _, err := client.ChatCompletion(ctx, req)
    if err != nil {
        if strings.Contains(err.Error(), "401") {
            return fmt.Errorf("invalid API key - please check your key at https://www.holysheep.ai/register")
        }
        return fmt.Errorf("API validation failed: %w", err)
    }
    
    return nil
}

2. ConnectionError: timeout / i/o timeout

สาเหตุ: เครือข่ายช้าหรือ API server ตอบสนองช้า

วิธีแก้ไข:

// เพิ่ม timeout ที่เหมาะสมและ retry logic
type ClientConfig struct {
    Timeout       time.Duration
    MaxRetries    int
    RetryDelay    time.Duration
}

func CreateRobustClient(config ClientConfig) *HolySheepClient {
    return &HolySheepClient{
        config: Config{
            BaseURL: "https://api.holysheep.ai/v1",
            APIKey:  os.Getenv("HOLYSHEEP_API_KEY"),
            Timeout: config.Timeout,
        },
        client: &http.Client{
            Timeout: config.Timeout,
            Transport: &http.Transport{
                DialContext: (&net.Dialer{
                    Timeout:   30 * time.Second,
                    KeepAlive: 30 * time.Second,
                }).DialContext,
                MaxIdleConns:        100,
                MaxIdleConnsPerHost: 100,
                IdleConnTimeout:     90 * time.Second,
            },
        },
    }
}

// ใช้งาน
func robustAPICall(ctx context.Context, client *HolySheepClient, req ChatRequest) (*ChatResponse, error) {
    maxRetries := 3
    retryDelay := 5 * time.Second
    
    for attempt := 0; attempt < maxRetries; attempt++ {
        if attempt > 0 {
            fmt.Printf("Retry attempt %d/%d after %v...\n", attempt, maxRetries, retryDelay)
            select {
            case <-time.After(retryDelay):
                retryDelay *= 2 // Exponential backoff
            case <-ctx.Done():
                return nil, ctx.Err()
            }
        }
        
        resp, err := client.ChatCompletion(ctx, req)
        if err == nil {
            return resp, nil
        }
        
        // ถ้าเป็น timeout error ให้ retry
        if strings.Contains(err.Error(), "timeout") || 
           strings.Contains(err.Error(), "i/o timeout") {
            continue
        }
        
        // Error อื่นๆ ไม่ต้อง retry
        return nil, err
    }
    
    return nil, fmt.Errorf("failed after %d retries", maxRetries)
}

3. context deadline exceeded

สาเหตุ: Context timeout หมดก่อนที่ request จะเสร็จ

วิธีแก้ไข:

// ใช้ context ที่มี timeout เหมาะสม
func processWithTimeout(ctx context.Context, requests []Request) {
    // คำนวณ timeout ที่เหมาะสม
    // สมมติว่าแต่ละ request ใช้เวลาประมาณ 2 วินาที + buffer 20%
    estimatedTime := time.Duration(len(requests)) * 2400 * time.Millisecond
    
    // เพิ่ม buffer 50% สำหรับ contingency
    totalTimeout := estimatedTime + (estimatedTime / 2)
    
    // ใช้ timeout ที่คำนวณได้ หรือ minimum 1 ชั่วโมง
    ctx, cancel := context.WithTimeout(ctx, totalTimeout)
    if ctx.Err() == context.DeadlineExceeded && totalTimeout < time.Hour {
        // ถ้าคำนวณได้น้อยกว่า 1 ชั่วโมง ให้ใช้ 1 ชั่วโมงแทน
        ctx, cancel = context.WithTimeout(context.Background(), time.Hour)
    }
    defer cancel()
    
    processor := NewConcurrentProcessor(os.Getenv("HOLYSHEEP_API_KEY"), 50, 100)
    results := processor.ProcessBatch(ctx, requests)
    
    // ตรวจสอบว่า context หมดหรือไม่
    if ctx.Err() == context.DeadlineExceeded {
        fmt.Printf("WARNING: Processing timed out! Processed %d/%d requests\n", 
            len(results), len(requests))
    }
}

4. 429 Too Many Requests / Rate Limit

สาเหตุ: เรียก API บ่อยเกินไปเกิน rate limit ของ API provider

วิธีแก้ไข:

// ใช้ rate limiter ที่ฉลาด
type AdaptiveRateLimiter struct {
    limiter     *rate.Limiter
    mu          sync.Mutex
    currentRate float64
    maxRate     float64
}

func NewAdaptiveRateLimiter(requestsPerSecond float64) *AdaptiveRateLimiter {
    return &AdaptiveRateLimiter{
        limiter:     rate.NewLimiter(rate.Limit(requestsPerSecond), 10),
        currentRate: requestsPerSecond,
        maxRate:     requestsPerSecond * 2,
    }
}

// WaitWithRetry รอจนกว่าจะเรียกได้ พร้อมปรับ rate อัตโนมัติ
func (a *AdaptiveRateLimiter) WaitWithRetry(ctx context.Context) error {
    for {
        err := a.limiter.Wait(ctx)
        if err == nil {
            return nil
        }
        
        // ถ้าเป็น rate limit error ให้ลด rate ลง
        a.mu.Lock()
        a.currentRate *= 0.8 // ลด 20%
        if a.currentRate < 1 {
            a.currentRate = 1
        }
        a.limiter.SetLimit(rate.Limit(a.currentRate))
        fmt.Printf("Rate limit hit - reducing to %.1f req/s\n", a.currentRate)
        a.mu.Unlock()
        
        // รอ 1 วินาทีก่อนลองใหม่
        select {
        case <-time.After(time.Second):
        case <-ctx.Done():
            return ctx.Err()
        }
    }
}

สรุป

การใช้ goroutine เพื่อเรียก AI API แบบ high concurrency สามารถเพิ่ม throughput ได้อย่างมาก แต่ต้องระวังเรื่อง:

ด้วยโค้ดในบทความนี้ ผมสามารถประมวลผล 10,000 requests ภายใน 10-15 นาที แทนที่จะใช้เวลา 5-6 ชั่วโมงแบบ sequential เร็วขึ้นเกือบ 30 เท่า!

อย่าลืมว่า HolySheep AI มี latency เฉลี่ยต่ำกว่า 50ms และราคาประหยัดมาก ลองนำไปใช้ดูนะครับ

👉 สมัคร HolySheep AI — รับเครดิตฟรีเมื่อลงทะเบียน

แหล่งข้อมูลที่เกี่ยวข้อง

บทความที่เกี่ยวข้อง