批量API设计：高效处理批量请求

可视化演示

动态展示RSA非对称加密算法的公钥加密、私钥解密全流程，直观呈现密钥生成、模幂运算和大数分解的核心原理。

认识批量API

假设你是一家礼品店的老板，早上要给8位会员送礼品。如果每送一份礼品都单独叫一次快递、单独填单、单独付运费，那一上午都会耗在“跑腿”上。批量API的思路就是把8份礼品装进一个大快递箱，填一张总单，一次寄出。至于快递公司内部怎么拆箱、分拣、派送，那是它的事——你只需完成一次交接。

在API设计里，一次“单件快递”就是一个HTTP请求。客户端发起一次请求，服务端返回一次响应，这就是一次网络往返。当我们需要处理N个彼此独立的子请求——比如给8张图片分别做分类——逐个发送就会产生N次往返。批量API把这个数量压缩到1次：客户端把N个子请求装进一个请求体，服务端统一受理后，把N个结果一起返回。

批量API不是简单的语法糖，它有实实在在的收益。第一，网络往返次数从N降到1，在高延迟网络下提升显著；第二，服务端可以合并身份校验、数据库连接等公共开销；第三，在AI推理等高吞吐场景下，批量请求能更充分地利用稀缺的GPU算力。

回到礼品店的例子。8位顾客意味着8次来回跑腿，而批量方式只需1次；网络延迟L就是每次跑腿的固定成本，无论箱子里有多少礼品，这笔路费只付一次。这就是批量API最核心的直觉：能一趟办完的事，不要跑N趟。

接口设计要点

快递大箱子里装着多件小礼品，每件小礼品都有自己的小标签，写着收件人、地址、物品名。如果你只是把礼品倒进箱子而不贴标签，分拣中心必然乱套。批量API的请求体也一样，必须为每个子请求提供足够的信息，才能让服务器“分拣”得清清楚楚。

请求体结构通常是一个数组，每个元素包含：唯一标识id（用于结果映射）、请求方法、路由路径、请求体内容。一个典型的AI分类批量请求长这样：

响应体则返回一个results数组，每个元素对应一个子请求的结果，包含id、status、data或error。这里最关键的设计是“id映射”：子请求带上唯一id，服务器在结果里原样返回这个id，客户端才能准确知道哪条结果属于哪个请求。虽然也可以按数组下标对应，但id方式更稳健，避免了增删顺序导致错位。URL设计方面，通用批量接口常用POST /api/v1/batch，针对单资源的批量操作可用POST /api/v1/items/batch。

大箱子外的总单子和每件礼品上的小标签各司其职：批量API的请求体是“打包清单”，响应体是“拆包回执”。设计清晰的接口结构，后续的扩展、排查和部分失败定位都会事半功倍。

{  "requests": [    {"id": "r1", "method": "POST", "path": "/classify", "body": {"image": "a.jpg"}},    {"id": "r2", "method": "POST", "path": "/classify", "body": {"image": "b.jpg"}}  ]}

部分成功处理

大箱子到了分拣中心，拆包时发现两件礼品因运输颠簸破损了。分拣员不会把其余6件完好的礼品也扔掉，而是把破损的两件单独登记为“异常件”，完好的继续派送。这是快递行业里最自然不过的事，却是批量API新手最容易踩坑的地方。

批量请求中，N个子请求共享一次HTTP连接，但它们的执行彼此独立。有的成功、有的失败是常态。错误隔离的含义正是：单条子请求的失败不能影响其他子请求的执行和返回。服务端要为每个子请求单独记录执行状态，而不是“一票否决”。

HTTP状态码如何设计？不建议用500代表“批量中有人失败”，那会丢失所有粒度信息。通用做法是：整体结构返回200，在响应体的每个子项里放置该子请求自己的status；想更规范可以使用207 Multi-Status。客户端根据每个子项的status决定是重试、降级还是直接展示错误。

“破损登记、完好的照常派送”——错误隔离让批量请求不至于“一损俱损”。对AI推理服务尤其如此：10张图里1张模糊导致识别失败，不应耽误其余9张的出图。

批量限制策略

快递公司规定每个箱子限重30公斤，并非不近人情，而是因为超重会带来一连串问题：搬运工搬不动、运输车超载、中转站爆仓。批量API同样必须设置“限重”，否则一个请求携带10万个子请求，服务端内存立刻告急，响应迟迟不返回，最终整体超时。

批量大小上限是最常见的限制，一般由服务端硬性规定，比如“单批最多100个”。超过上限直接返回4xx错误，而不是默默截断。上限的取值需要结合业务调整：AI推理请求体大、耗时长，上限应设得小一些；简单查询类请求体量小，上限可以放宽。

除了大小上限，还要做并发控制：限制同一时刻正在处理的批次数，防止客户端疯狂提交超大批次打垮后端。配套的限流与熔断是安全双保险——当错误率超过阈值时快速拒绝新批次，让服务先缓过来。

限重规则不是“不友好”，恰恰是保护整个快递网络不瘫痪。设计批量API时，让上限可配置、可监控，远比一味追求“大包裹”更专业。

性能优化方法

快递分拣中心之所以高效，靠的不是某一个快递员拼命跑，而是一套自动化流水线：自动扫码、传送带分流、分区域装车。批量API的性能优化同样需要这种“流水线思维”。

吞吐量提升的核心是让服务端并行处理同一批次内的子请求。假设服务端有8个worker线程，一个批次内有8个子请求，并行处理后单批耗时接近一个请求的耗时，吞吐量能提升数倍；如果串行处理，大批次反而会更慢。AI推理场景尤其明显：GPU本身擅长批处理，把多个推理请求打包在一起送入显存，计算效率远高于逐个发送。

网络效率方面，批量请求天然减少了HTTP头部、TLS握手等固定开销。超时策略也要专门设计：为整个批次设置总超时，同时为每个子请求设置内部超时，避免一个慢请求拖垮整批结果。实践上常用“整体超时 + 子任务截止时间”的双层结构。

流水线思维的核心是“把资源集中在批量粒度上调度”。一个优秀的批量API，吞吐量应当随着批量大小和并发度近似线性增长，直到触达上限——就像分拣中心在满负荷之前，传送带越快，出件越多。

工程实践落地

一家成熟的快递公司不会只盯着“今天发了多少件”，还会盯着破损率、延误率、各网点吞吐情况。批量API上线后同样需要一套运营体系，否则问题会藏在“平均耗时”的假象里。

版本兼容方面，批量接口一旦发布就意味着长期承诺。新增子请求字段要向后兼容，必要时通过版本号优雅升级，避免破坏存量客户端。监控告警方面，除了传统的接口成功率，还要监控“批次内子失败率”——即使最外层HTTP返回200，也要能发现部分子请求持续失败并触发告警。安全防护方面，批量接口是攻击者的“放大镜”：一次恶意请求可携带成千上万个子任务。必须做鉴权校验、配额控制和请求体大小限制，防止批量接口被用来发起放大攻击。

这里还有一个反直觉但重要的工程事实：批量越大，整批至少出现一次失败的概率越高。

从“能跑”到“跑得稳”，靠的是版本、监控、安全这三根支柱。就像快递公司对每个包裹全程可追溯，你对每个批次的每个子请求也要“看得见、管得住”。

AI 应用场景

Python 演示

import randomimport time  class BatchResult:    """单个子请求的处理结果"""    def __init__(self, request_id, status, data=None, error=""):        self.request_id = request_id        self.status = status        self.data = data        self.error = error  class BatchProcessor:    """模拟批量API的服务端处理器"""     def __init__(self, max_batch_size=10):        self.max_batch_size = max_batch_size        self.total_count = 0        self.failed_count = 0     def process_batch(self, requests):        """处理一个批量请求，返回每个子请求的结果"""        if len(requests) > self.max_batch_size:            raise ValueError("批量大小超过上限，请拆分后重试")        results = []        for req in requests:            result = self._process_one(req)            results.append(result)            self.total_count += 1            if result.status != 200:                self.failed_count += 1        return results     def _process_one(self, req):        """模拟单条请求处理，随机展示部分失败"""        time.sleep(0.005)        if random.random() < 0.2:            return BatchResult(req["id"], 503, error="服务繁忙")        return BatchResult(req["id"], 200, data="推理完成")  def send_batch(processor, requests):    """客户端调用批量API的入口"""    try:        results = processor.process_batch(requests)        success = sum(1 for r in results if r.status == 200)        print(f"批量请求完成: 成功 {success}/{len(results)}")        for r in results:            if r.status != 200:                print(f"  子请求 {r.request_id} 失败: {r.error}")    except ValueError as exc:        print(f"批量请求被拒绝: {exc}")  if __name__ == "__main__":    processor = BatchProcessor(max_batch_size=5)    batch = [{"id": i, "task": "文本分类"} for i in range(1, 6)]    send_batch(processor, batch)    big_batch = [{"id": i, "task": "文本分类"} for i in range(6)]    send_batch(processor, big_batch)

1. if len(requests) > self.max_batch_size:批量大小限制：超过上限直接拒绝整批，与快递限重同理。
2. results.append(result)把每个子请求的结果单独收集，支撑部分成功返回。
3. if result.status != 200: self.failed_count += 1失败只计数不中断，这就是错误隔离。
4. raise ValueError("批量大小超过上限，请拆分后重试")超限时给出明确提示，而不是静默截断。
5. send_batch(processor, big_batch)第二个调用故意超过上限，演示服务端如何拒收超重包裹。

常见误区

面试题