动态批处理（Dynamic Batching）原理与实现：破解高并发系统吞吐瓶颈

“我接手过不少服务端的性能问题，最后都绕到了同一类模式上：单个请求本身不难处理，难的是请求量一上来，系统的吞吐就卡在一个完全不成比例的低水位。动态批处理（Dynamic Batching）就是专门用来打破这种局面的策略之一：在并发请求到达时，不急着挨个处理，而是攒成一个批次，一次性交给下游执行，再把结果拆开返回给各自的调用方。适合读这篇文章的人很明确，正在做模型推理服务、数据库写入层、RPC 转发层，或者任何“单次固定开销很高，批量处理能摊薄固定开销”的系统。读完你能知道它内部到底怎么运作，能直接照抄一套能跑的实现，也能搞明白参数到底该怎么调。”

1. 并发场景下批处理的困境与破解思路

1.1 固定批处理的痛点

很多团队第一次接触批处理，都会从“固定窗口”或者“固定大小”入手。比如“每 20 毫秒把队列里攒到的请求全部处理掉”，或者“攒够 32 个请求再处理”。这两种方案实现都非常简单，可一旦碰到真实流量，问题就来了。

固定窗口的方案，在低峰期很吃亏。假设窗口是 20 毫秒，某个时段每秒只有 50 个请求，平均每个窗口只能攒到 1 个请求。为了等窗口结束，每个请求都要白白多等 10 到 20 毫秒，可等到最后批大小还是 1，批处理等于没有发生，延迟倒是实打实加了一截。固定大小的方案正好反过来，它在低峰期表现还不错，攒够数量就立刻处理，可一旦高峰期流量短时间内激增，队列里可能瞬间排了几百个请求，攒够 32 个就发一批，看起来挺合理，但实际上如果处理器本身的固定开销很低、单批容量可以做得更大，那这种“攒够即发”反而会让系统不敢把批次做大，白白浪费高吞吐的机会。

这里面的矛盾点，本质上是一个动态权衡：批大小和等待时间必须同时被纳入决策。流量低的时候，宁可少攒一点，也不能让调用方承受没有必要的高延迟；流量高的时候，上游积压的请求本来就多，这时候多等一小段时间，换来一批更满的请求，下游吞吐反而更高，平均排队时间更短。固定窗口和固定大小，都只抓住了其中一个维度。

1.2 动态批处理解决问题的核心逻辑

动态批处理的思路，是把“时间”和“数量”都当作触发条件，但不是简单地“或”关系，而是让它们形成一种互补：

• 第一个请求到达时，批处理器开始计时；
• 在计时窗口内，如果有新的请求到达，就继续吸收进当前批次；
• 如果批次数量已经达到容量上限，即使窗口时间还没到，也立刻触发处理；
• 如果窗口时间已经到了，不管当前攒了多少个请求，只要至少有一个请求，就触发处理。

这样，低峰期时，窗口到期后批里可能只有一两个请求，可以很快发出，避免无谓等待；高峰期时，请求源源不断涌进来，往往还没到窗口上限，批就已经攒满并触发了。最终的效果是：流量越低，批处理器表现得越像一个直接透传的转发层；流量越高，它越像一个主动聚合的批量工厂。这就是它叫做“Dynamic”的原因，行为会随着流量自动变化。

从更底层的经济学角度看，动态批处理做了两件事：第一，摊薄固定开销。远程调用、模型推理、数据库事务，都存在与业务本身无关的固定成本，比如网络 RTT、序列化、线程切换、GPU 内核启动，批量越大，每个请求分摊的成本越小。第二，减少下游热点。下游如果是一个有并发上限的资源，比如数据库连接池或 GPU 执行队列，上游将请求合并后再访问，能把下游的并发压力降低一个数量级，从根上避免排队风暴。

不过必须说清楚，动态批处理不是万能的，它只适合那些“下游可以一次性接收一批输入、并返回一批输出”的处理器。如果下游本身是一个线性交互的 API，比如每个请求都必须单独鉴权、单独记录，那批量处理后端要么不支持、要么收益很小，这时候强行引入批处理反而增加复杂度。

2. 动态批处理的整体设计与运行机制

2.1 核心要素：窗口、批量上限与最小批量

一个标准的动态批处理器，需要三个核心参数：

第一个是max_batch_size，代表单批最大能容纳的请求数。这个上限通常不是由批处理器决定的，而是由下游能力决定的。拿 GPU 推理举例，显存能同时容纳的样本数量就是一个硬上限，超过这个值要么显存溢出，要么延迟恶化到不可接受。数据库批量插入同理，单条 SQL 能承载的 VALUES 数量、事务日志单次写入的长度，都限制批次大小。所以max_batch_size应该由下游评测得出，而不是凭感觉定。

第二个是window，也就是最大等待窗口。它是单请求愿意为“凑批”支付的额外延迟上限。设置时需要非常谨慎，因为它直接贡献到接口的延迟里。假设你的服务承诺 P99 延迟 100 毫秒，批量执行本身需要 30 毫秒，那么窗口最多只能给到 50 毫秒左右，剩下还要留网络和排队余量。实际项目中，窗口经常设置在 5 到 30 毫秒之间，既不太影响用户体验，又能攒出可观的批次。

第三个是min_batch_size，通常在实际系统中是 1，但也可以根据情况提高。比如有一类下游处理器，批量处理 1 个请求和批量处理 5 个请求的耗时几乎一样，那你就不值得为了 1 个请求提前触发，可以设置min_batch_size=5，告诉批处理器“至少攒 5 个才算有效批次”。代价是低峰期的单请求延迟会被拉高，所以这个值一般只在业务可以容忍时使用。我见过一些团队把它做成动态配置，高峰期调高、低峰期调低，效果非常好。

这几个参数的关系，可以类比成收费站的蓄车策略：max_batch_size是一条车道最多能停多少辆车，window是栏杆最多等多久必须放行，min_batch_size是如果车太少，宁可多等一会儿也不值得专门开一次闸。

2.2 请求积压与响应的路由：队列加事件

动态批处理在并发场景下，实际上是在多个调用线程和一个后台工作线程之间，搭了一条数据管道。调用线程把请求丢进队列后，不能傻等，而是要有一个自己专属的信令通道，让工作线程在批量处理完成后，能精确地把结果送回给对应的调用线程。这个设计要特别注意，不能直接在队列里放原始请求，必须用“占位对象”包装。

占位对象里至少包含三样东西：请求体、存放结果的槽位、一个用于通知完成的事件对象。调用线程submit时，把自己的占位对象放进公共队列；后台线程把一批占位对象取出来，调用下游处理器；处理完后，按顺序把结果写入每个占位对象的结果槽，然后触发事件。调用线程在submit里等待这个事件，事件一旦被触发，立刻读取结果并返回。

这个机制要解决的问题是“多对多”：多个调用线程并发提交，可能交错进入多个批次；而一个批次又包含多个调用者的请求。用房间号加门铃来类比最直观：每个调用者拿到一个房间门牌号，进了一个批次，相当于和一屋子人住了同一间酒店，但彼此不干扰。批量处理器就是那个服务员，统一给整层楼送餐，敲谁的房门，谁就开门拿自己的结果。

线程安全方面，公共队列必须加锁保护，并且在入队时通知等待的工作线程。等待操作不能占着锁，否则其他调用线程无法入队，批处理就退化成串行。通常使用条件变量，入队后notify唤醒工作线程，工作线程在无请求时wait释放锁。这里有一个典型误区：新手容易在wait之后立刻处理队列，却忽略了 wait 被超时唤醒和真条件满足之间的差别，必须用while循环再检查一次条件，防止虚假唤醒。

3. 实操：多线程与异步两种完整实现

3.1 同步动态批处理器：多线程版

先给一套可以直接落地的多线程实现，适用于大多数后端服务。它对外暴露一个submit方法，调用方传入请求体，阻塞等待返回结果；内部由一个线程负责攒批、执行批量处理、分发结果。

import threading import time from collections import deque from typing import Any, Callable, List, Optional class DynamicBatcher: def __init__( self, processor: Callable[[List[Any]], List[Any]], max_batch_size: int = 32, window: float = 0.02, min_batch_size: int = 1, ): self._processor = processor self._max_batch_size = max_batch_size self._window = window self._min_batch_size = min_batch_size self._lock = threading.Lock() self._cond = threading.Condition(self._lock) self._pending: deque = deque() self._closed = False self._worker = threading.Thread(target=self._run, daemon=True) self._worker.start() def submit(self, payload: Any, timeout: float = 10.0) -> Any: item = { "payload": payload, "result": None, "error": None, "done": threading.Event(), } with self._lock: if self._closed: raise RuntimeError("batcher is closed") self._pending.append(item) self._cond.notify() if not item["done"].wait(timeout): raise TimeoutError( "batch wait timeout, payload={!r}".format(payload) ) if item["error"] is not None: raise item["error"] return item["result"] def close(self) -> None: with self._lock: self._closed = True self._cond.notify_all() self._worker.join(timeout=5) def _run(self) -> None: while not self._closed: batch = self._collect() if not batch: continue requests = [item["payload"] for item in batch] try: responses = self._processor(requests) if len(responses) != len(requests): raise RuntimeError( "processor returned {} results, expected {}".format( len(responses), len(requests) ) ) for item, resp in zip(batch, responses): item["result"] = resp except Exception as exc: for item in batch: item["error"] = exc finally: for item in batch: item["done"].set() def _collect(self) -> list: with self._lock: while not self._pending and not self._closed: self._cond.wait() batch = [] deadline = time.monotonic() + self._window while len(batch) < self._max_batch_size: if self._pending: batch.append(self._pending.popleft()) if len(batch) >= self._max_batch_size: break continue if not batch: # 队列为空，说明刚刚处理完上一批，进入下一轮等待 continue remaining = deadline - time.monotonic() if remaining <= 0 or len(batch) >= self._min_batch_size: break self._cond.wait(remaining) return batch

这个实现比较简洁，但里面有三个地方需要重点解释。

第一个是_collect的核心循环。它的行为逻辑是：只要队列里有请求，就尽量吸收进当前批次；一旦批次满了，立即返回。如果队列空了，就判断是否已到达窗口截止时间，或者当前请求数是否已达到min_batch_size，满足任一条件就结束本次收集。这样做的好处是，在低峰期如果第一个请求到达后，队列里第二、第三个请求很快跟上来，就可以在较短时间内攒出一个有用的批次；如果一直没人来，窗口也不会无限拖下去。

第二个是异常处理。批量处理器一旦抛异常，我没有选择让整个批次的调用方全部跟着崩溃，而是把异常对象逐个填入每个请求的错误槽。这样调用方会收到异常，但至少不会卡死。如果调用方在高并发下出现大规模失败，也能通过错误信息快速定位到是处理器的问题，而不是批处理器本身的问题。

第三个是submit的超时逻辑。调用方等待事件时设置了超时，如果超时发生，说明这批请求可能因为下游堆积而迟迟没有执行。注意这时不能从公共队列里移除该项，因为它可能已经进了工作线程的本地批次里，队列中根本找不到它。让结果被写进一个没人读取的占位对象，是安全的，唯一的代价是那部分处理能力被浪费了。

3.2 异步动态批处理器：asyncio 版

在高性能网络服务里，线程模型有时并不是首选。尤其是调用方本身跑在事件循环上时，如果让它们在线程里wait，会白白占用系统线程。异步版本的动态批处理器更适合这种情况。

import asyncio import time from typing import Any, Awaitable, Callable, List class AsyncDynamicBatcher: def __init__( self, processor: Callable[[List[Any]], Awaitable[List[Any]]], max_batch_size: int = 32, window: float = 0.02, min_batch_size: int = 1, ): self._processor = processor self._max_batch_size = max_batch_size self._window = window self._min_batch_size = min_batch_size self._lock = asyncio.Lock() self._cond = asyncio.Condition(self._lock) self._pending: list = [] self._closed = False self._task = asyncio.create_task(self._run()) async def submit(self, payload: Any, timeout: float = 10.0) -> Any: fut = asyncio.get_running_loop().create_future() async with self._lock: if self._closed: raise RuntimeError("batcher is closed") self._pending.append((payload, fut)) self._cond.notify() try: return await asyncio.wait_for(fut, timeout) except asyncio.TimeoutError: fut.cancel() raise TimeoutError( "batch wait timeout, payload={!r}".format(payload) ) async def close(self) -> None: async with self._lock: self._closed = True self._cond.notify_all() if self._task: await self._task async def _run(self) -> None: while not self._closed: batch = await self._collect() if not batch: continue payloads = [item[0] for item in batch] try: responses = await self._processor(payloads) if len(responses) != len(payloads): raise RuntimeError( "processor returned {} results, expected {}".format( len(responses), len(payloads) ) ) except Exception as exc: responses = [exc] * len(payloads) for (_, fut), resp in zip(batch, responses): if not fut.done(): fut.set_result(resp) async def _collect(self) -> list: async with self._lock: while not self._pending and not self._closed: await self._cond.wait() batch = [] deadline = time.monotonic() + self._window while len(batch) < self._max_batch_size: if self._pending: batch.append(self._pending.pop(0)) if len(batch) >= self._max_batch_size: break continue if not batch: continue remaining = deadline - time.monotonic() if remaining <= 0 or len(batch) >= self._min_batch_size: break await self._cond.wait(remaining) return batch

异步版本在结构上跟多线程版本几乎一一对应，但有两个关键差异。

第一，未来的容器是asyncio.Future，它在事件循环上等待完成，不占系统线程。调用方在await submit期间，事件循环还可以处理其他连接，整体并发能力要比多线程模型高很多。第二，批量处理器本身也被设计成协程，因为模型推理或数据库读取这类操作，如果是异步客户端，用await等结果就不会阻塞事件循环。如果下游没有异步客户端，可以先用asyncio.to_thread把同步调用包一层，让事件循环在等待期间继续服务别人。

这两段实现的完整代码可以直接抄进实际项目。要注意类里的pop(0)在请求量极高时会有列表头部删除的 O(n) 开销，如果要上大规模生产，可以把_pending改成collections.deque，或者用一个索引指针模拟消费队列，避免频繁移位。这是个典型的实现细节坑，不刻意提一下可能很多人会忽略。

3.3 参数推演与实际选型建议

光有代码还不够，关键是怎么给你的业务定一套靠谱的初始参数。我来推演一个实际例子。

假设你负责一个模型推理服务，单次单条请求推理耗时 8 毫秒，GPU 资源充足，但显存限制单批最多 64 个样本。接口要求 P99 延迟不超过 100 毫秒。你初步设定max_batch_size=64，window=20ms。这时候可以估算极端情况下的延迟构成：网络和排队保守 20 毫秒，窗口 20 毫秒，批量执行时间取决于实际批大小。如果批大小接近 64，单批推理耗时可能增长到 30 到 50 毫秒，加起来大概 90 毫秒，刚好卡在 P99 边缘。这个参数风险偏大，更稳的初始方案是max_batch_size=32、window=10ms，给其他波动留出余量。

流量的影响可以用一个简易模型理解：平均批大小约等于QPS * window，取整后不能超过max_batch_size。假设窗口 10 毫秒，每秒请求数 1000，那么每个窗口理论上平均会来 10 个请求。GPU 一次处理 10 个样本，比一次处理 1 个样本在整体吞吐上要高很多。如果 QPS 涨到 5000，每个窗口会来 50 个请求，接近 64 的上限，说明此时批基本是满载的，吞吐已经逼近硬上限，再提升只能靠增大max_batch_size或扩展资源。

实际选型时，我会给一个非常朴素的建议：先按max_batch_size = 下游安全上限的一半起步，窗口从 10 毫秒开始，然后通过压测逐步调整。压测时重点看两个指标：一个是单请求延迟的 P99，另一个是单位时间成功处理的请求数 QPS。每次调整只改一个参数，对比前后数据，不要同时动两个变量，否则根本说不清是哪个参数带来的变化。

4. 常见问题与排查技巧实录

4.1 容易踩的坑汇总

我在多个项目里帮人排查过批处理相关的问题，有些坑出现的频率非常高，整理在这里，省得大家再走一遍弯路。

第一个坑是持锁调用下游处理器。有人觉得队列取出来后，顺手在锁里执行批量处理逻辑很省事。但实际上，如果处理时间较长，所有调用线程的submit都会卡在抢锁上，系统直接退化成完全串行，批处理比不批处理还慢。正确做法是加锁只做队列操作，取完批之后立即释放锁，再执行下游调用。

第二个坑是处理结果数量不匹配。有些下游接口会出于各种原因少返回几条结果，比如数据库批量插入只返回影响行数而不返回详细结果。如果代码里不做长度校验，zip会静默丢弃后面的请求，导致部分调用方永远等不到事件，直到超时。我在实现里特意加了一个长度校验，不匹配就整批报错，宁可让所有调用方看到异常，也不能让谁无声无息地卡死。

第三个坑是伪唤醒。线程条件变量的wait可能在没有主动 notify 时被系统唤醒，或者多个工作线程同时被唤醒后只有一个能抢到任务。如果使用if而非while检查队列条件，就会在队列为空时继续走下去，取到空批次，直接跳过处理。这个 bug 的可怕之处在于它不是必现的，只有高负载或信号竞争时才会偶发，极难排查。所以攒批循环必须用while重新验证队列状态。

第四个坑是超时和批次并发写入同一个结果槽。调用方在等待超时后，可能会直接销毁请求上下文，但工作线程仍然在处理这个请求。如果结果槽指向的是可变对象，可能发生一边写一边读的并发问题。我推荐的方案是让每个调用方持有自己的占位对象，超时后不再读取。若确实需要在超时后复用请求体，必须在占位对象里加锁或使用不可变结果。

下面是一个快速排查速查表，可以直接打印出来贴在工位上：

4.2 动态调整批量的进阶经验

跑稳基础版本之后，可以进一步做自适应动态批处理。标准动态批处理已经能够根据流量自动变化，但在某些场景下，还想要更激进一点，让批处理器感知当下的系统压力，动态调整窗口长度。

我实践过一种简单的方案：工作线程在每次批次结束后，统计当批的实际大小和积压请求数。如果积压一直很多，说明窗口可以适当放长，因为调用方反正在排队，多等几毫秒换更高的吞吐是值得的。如果积压很少、批大小长期为 1，说明窗口需要收短，否则每个请求都在无谓地等窗口。用公式表达就是：

new_window = base_window * (1 + k * backlog_factor) * (backlog_factor - batch_utilization)

严格说这个公式本身不是重点，真正重要的是设计思路：窗口长度应该是系统压力的函数，而不是一个恒定常量。很多开源框架里的minimum_window、maximum_window、preferred_batch_size参数，本质上都是在表达这种自适应策略。我建议把这部分做成可配置项，配合监控面板实时调整，会比拍脑袋定参数可靠得多。

另外一个很实用的小技巧，是针对特殊请求的绕过通道。不是所有请求都适合进批处理，比如部分实时交互请求对延迟极其敏感，而普通异步任务可以容忍多等几十毫秒。可以设计成：批处理器内部维护两个队列，一个普通队列走动态批处理，一个紧急队列每批优先合并处理，或者直接透传。实现成本不高，但对用户体验的提升非常明显。这就是“动态”的另一种含义，不仅仅是批量大小动态，连入队策略本身也是动态的。

最后再说一个关于监控的建议：批处理系统一定要暴露几个核心指标，当前积压请求数、最近窗口平均批大小、每批的平均执行耗时、提交到返回的平均总延迟。这四个指标能让你在线上问题发生的几分钟内判断出问题到底出在攒批逻辑还是下游处理器。而我排查过的绝大多数批处理故障，光是看“平均批大小”这一个指标就能定位一半以上的原因。把这套东西沉淀成标准指标，后面无论接入什么业务，都能用同一套方法论快速排障。