Golang中如何高效实现数千个API请求的并行处理

在Golang中高效处理数千个API请求的并行处理，可以使用goroutine和channel结合worker pool模式。以下是具体实现方案：

package main

import (
    "encoding/json"
    "fmt"
    "sync"
    "time"
    "net/http"
    "strconv"
    "log"
)

const (
    totalItems     = 5000
    pageSize       = 500
    maxConcurrency = 50 // 控制并发数，避免过多连接
)

type Product struct {
    ID       int     `json:"id"`
    Name     string  `json:"name"`
    Price    float64 `json:"price"`
    Category string  `json:"category"`
}

type APIResponse struct {
    Products []Product `json:"products"`
    Offset   int       `json:"offset"`
    Total    int       `json:"total"`
}

func fetchPage(offset int, client *http.Client) ([]Product, error) {
    url := fmt.Sprintf("https://api.example.com/products?limit=%d&offset=%d", pageSize, offset)
    
    resp, err := client.Get(url)
    if err != nil {
        return nil, err
    }
    defer resp.Body.Close()
    
    var apiResp APIResponse
    if err := json.NewDecoder(resp.Body).Decode(&apiResp); err != nil {
        return nil, err
    }
    
    return apiResp.Products, nil
}

func worker(id int, jobs <-chan int, results chan<- []Product, wg *sync.WaitGroup, client *http.Client) {
    defer wg.Done()
    
    for offset := range jobs {
        log.Printf("Worker %d fetching offset %d", id, offset)
        
        products, err := fetchPage(offset, client)
        if err != nil {
            log.Printf("Worker %d error fetching offset %d: %v", id, offset, err)
            results <- []Product{} // 发送空切片保持通道同步
            continue
        }
        
        results <- products
    }
}

func main() {
    start := time.Now()
    
    // 计算需要的请求数量
    numRequests := (totalItems + pageSize - 1) / pageSize
    
    // 创建带缓冲的channel
    jobs := make(chan int, numRequests)
    results := make(chan []Product, numRequests)
    
    // 创建HTTP客户端（复用连接）
    client := &http.Client{
        Timeout: 30 * time.Second,
        Transport: &http.Transport{
            MaxIdleConns:        maxConcurrency,
            MaxIdleConnsPerHost: maxConcurrency,
            IdleConnTimeout:     90 * time.Second,
        },
    }
    
    var wg sync.WaitGroup
    
    // 启动worker pool
    for w := 1; w <= maxConcurrency; w++ {
        wg.Add(1)
        go worker(w, jobs, results, &wg, client)
    }
    
    // 发送任务到jobs channel
    for i := 0; i < numRequests; i++ {
        offset := i * pageSize
        jobs <- offset
    }
    close(jobs)
    
    // 等待所有worker完成
    go func() {
        wg.Wait()
        close(results)
    }()
    
    // 收集结果
    var allProducts []Product
    var mu sync.Mutex
    
    // 使用多个goroutine处理结果
    var resultWg sync.WaitGroup
    resultWg.Add(1)
    
    go func() {
        defer resultWg.Done()
        
        for products := range results {
            if len(products) > 0 {
                mu.Lock()
                allProducts = append(allProducts, products...)
                mu.Unlock()
            }
        }
    }()
    
    resultWg.Wait()
    
    // 输出统计信息
    elapsed := time.Since(start)
    log.Printf("Fetched %d products in %v", len(allProducts), elapsed)
    log.Printf("Average: %.2f products/second", float64(len(allProducts))/elapsed.Seconds())
    
    // 数据存储示例
    storeProducts(allProducts)
}

// 存储产品的示例函数
func storeProducts(products []Product) {
    // 使用map按类别分组
    productsByCategory := make(map[string][]Product)
    
    for _, p := range products {
        productsByCategory[p.Category] = append(productsByCategory[p.Category], p)
    }
    
    // 转换为JSON保存
    jsonData, err := json.MarshalIndent(productsByCategory, "", "  ")
    if err != nil {
        log.Printf("Error marshaling products: %v", err)
        return
    }
    
    // 这里可以保存到文件或数据库
    fmt.Printf("Total categories: %d\n", len(productsByCategory))
    
    // 示例：打印每个类别的产品数量
    for category, prods := range productsByCategory {
        fmt.Printf("Category %s: %d products\n", category, len(prods))
    }
    
    _ = jsonData // 实际使用中需要保存到文件
}

// 使用sync.Map处理并发写入的替代方案
func processWithSyncMap(products []Product) {
    var productMap sync.Map
    
    var wg sync.WaitGroup
    for _, p := range products {
        wg.Add(1)
        go func(product Product) {
            defer wg.Done()
            // 以ID为key存储产品
            productMap.Store(strconv.Itoa(product.ID), product)
        }(p)
    }
    wg.Wait()
    
    // 遍历sync.Map
    productMap.Range(func(key, value interface{}) bool {
        fmt.Printf("Product ID %s: %v\n", key, value)
        return true
    })
}

关键优化点：

1. Worker Pool模式：控制并发goroutine数量，避免创建5000+个goroutine
2. HTTP连接复用：使用自定义Transport配置连接池
3. 缓冲Channel：避免goroutine阻塞
4. 并发安全：使用sync.Mutex保护共享数据，或使用sync.Map
5. 错误处理：worker中单独处理错误，不影响其他请求

数据结构建议：

• 使用[]Product切片存储所有产品
• 使用map[string][]Product按类别分组
• 考虑使用sync.Map处理高并发读写

存储方案：

• 内存中处理：使用切片和map
• 持久化存储：转换为JSON保存到文件，或存入数据库
• 分批处理：每收集一定数量后批量保存