高并发场景下API接口批量调用工具的设计与实现

1次阅读

没有评论

共计 2467 个字符，预计需要花费 7 分钟才能阅读完成。

在微服务架构中，服务间通信主要依赖 API 调用。当系统规模扩大时，频繁的单次 API 调用会带来显著性能问题：

每次调用都需要建立 TCP 连接（即使使用 HTTP/1.1 Keep-Alive）
每个请求都有固定的协议开销（HTTP 头部等）
服务端需要为每个请求分配独立线程 / 协程资源
网络往返时间（RTT）成为主要延迟来源

实测数据显示，在 100TPS 的调用压力下，单次调用模式会导致：

服务端线程池快速耗尽
客户端出现大量 TIME_WAIT 连接
95% 的请求延迟超过 300ms

优点：单连接多请求、头部压缩
缺点：需要服务端支持、无法跨服务合并请求

优点：彻底异步化、削峰填谷
缺点：引入新组件、增加系统复杂度

优点：减少网络开销、提升吞吐量
缺点：需要客户端改造、响应延迟可能增加

采用双重触发机制确保及时处理：

时间窗口：最大等待 100ms
容量阈值：最多聚合 50 个请求

// Go 实现聚合器
type BatchAggregator struct {
    maxWait     time.Duration
    maxItems    int
    batchChan   chan Request
    flushChan   chan []Request}

func (a *BatchAggregator) run() {buffer := make([]Request, 0, a.maxItems)
    timer := time.NewTimer(a.maxWait)

    for {
        select {
        case req := <-a.batchChan:
            buffer = append(buffer, req)
            if len(buffer) >= a.maxItems {
                a.flushChan <- buffer
                buffer = buffer[:0]
                timer.Reset(a.maxWait)
            }
        case <-timer.C:
            if len(buffer) > 0 {
                a.flushChan <- buffer
                buffer = buffer[:0]
            }
            timer.Reset(a.maxWait)
        }
    }
}

为每个原始请求生成唯一 traceID
服务端返回批量响应时保持相同顺序
根据 traceID 将结果分发回原始调用方

# Python 异步处理示例
async def dispatch_requests(batch_requests):
    # 发送批量请求
    batch_response = await post_batch_api(batch_requests)

    # 拆分响应
    results = {}
    for i, req in enumerate(batch_requests):
        results[req.trace_id] = batch_response[i]

    return results

采用指数退避算法：

首次失败：立即重试
第二次失败：等待 200ms
第三次失败：等待 400ms
超过 3 次：标记为彻底失败

// 连接池管理
type ConnPool struct {
    pool    chan *http.Client
    maxSize int
}

func (p *ConnPool) Get() *http.Client {
    select {
    case conn := <-p.pool:
        return conn
    default:
        return &http.Client{Timeout: 30 * time.Second}
    }
}

// 熔断器实现
type CircuitBreaker struct {
    failureThreshold int
    resetTimeout     time.Duration
    lastFailure      time.Time
    failures         int
}

func (cb *CircuitBreaker) Allow() bool {
    if cb.failures >= cb.failureThreshold {return time.Since(cb.lastFailure) > cb.resetTimeout
    }
    return true
}

import aiohttp
from datetime import datetime

class BatchProcessor:
    def __init__(self):
        self.session = aiohttp.ClientSession()
        self.semaphore = asyncio.Semaphore(100)  # 并发控制

    async def process_batch(self, requests):
        async with self.semaphore:
            try:
                async with self.session.post(
                    'https://api.example.com/batch',
                    json=[r.to_dict() for r in requests]
                ) as resp:
                    return await resp.json()
            except Exception as e:
                self.retry_policy(requests)

测试环境：4 核 8G 云服务器，本地网络延迟 <2ms