4. 动态batching策略

说实话，静态batching虽然简单，但有个致命问题——它太死板了。你设个batch size=8，结果只来了3个请求，剩下5个位置就空着。GPU算力白白浪费，你说心疼不心疼？

动态batching就是来解决这个问题的。它的核心思想很简单：不等了，有多少来多少，凑够一波就发车。但这里面的门道，比你想的要深得多。

4.1 动态batching原理：从“等车”到“拼车”

我打个比方你就明白了。静态batching就像公交车，固定时间发车，不管车上几个人。动态batching呢，更像拼车——司机等一会儿，凑够人就出发，实在等不到人，时间到了也得走。

具体到技术层面，动态batching干了三件事：

• 请求收集：把一段时间内到达的请求攒起来
• 动态组批：根据当前请求量，实时决定batch大小
• 弹性调度：batch大小不是固定的，可以在1到max_batch之间浮动

这里有个关键点——padding。不同请求的输入长度不一样，组批时得把短的补长。我在项目中遇到过，如果padding策略没选好，内存浪费能到40%以上。后面我会讲怎么优化。

4.2 请求排队与超时机制：别让用户等到花儿都谢了

动态batching最头疼的问题是什么？排队策略。你想想看，请求来了，是先进先出？还是优先处理短请求？这里面学问大了。

我常用的排队策略有三种：

策略	原理	适用场景	我的评价
FIFO（先进先出）	谁先来谁先走	请求长度均匀	公平但效率一般
Shortest First	短请求优先	长度差异大	吞吐高，但长请求容易饿死
Binning（分桶）	按长度分桶，桶内FIFO	混合场景	我项目里最常用的方案

嗯，这里要注意——超时机制是动态batching的保底策略。我曾经踩过一个坑：有个服务把超时设成了5秒，结果高峰期请求排队时间飙到8秒，用户端全超时了。后来我改成动态超时，根据当前队列长度自动调整，才把问题解决。

4.3 实现一个简单的动态batching调度器

光说不练假把式。我给你写个简化版的调度器，核心逻辑就100行左右。你拿回去改改就能用。

import asyncio
import time
from collections import deque

class DynamicBatcher:
    def __init__(self, max_batch=8, max_wait=0.1, model_fn=None):
        self.max_batch = max_batch      # 最大batch大小
        self.max_wait = max_wait        # 最长等待时间（秒）
        self.model_fn = model_fn        # 模型推理函数
        self.queue = deque()            # 请求队列
        self._running = False
        
    async def submit(self, input_data):
        """提交一个请求，返回推理结果"""
        future = asyncio.Future()
        self.queue.append((input_data, future, time.time()))
        return await future
    
    async def _batch_processor(self):
        """后台批处理协程"""
        self._running = True
        while self._running:
            # 等待队列非空
            while len(self.queue) == 0:
                await asyncio.sleep(0.001)
            
            # 收集一批请求
            batch = []
            deadline = time.time() + self.max_wait
            
            while len(batch) < self.max_batch:
                if len(self.queue) == 0:
                    # 队列空了，等一会儿
                    await asyncio.sleep(0.001)
                    if time.time() > deadline:
                        break
                    continue
                
                # 取出队首请求
                input_data, future, arrival = self.queue.popleft()
                batch.append((input_data, future))
                
                # 检查是否超时
                if time.time() > deadline:
                    break
            
            # 执行推理
            if batch:
                inputs = [item[0] for item in batch]
                futures = [item[1] for item in batch]
                
                try:
                    # 这里做padding和推理
                    results = await self.model_fn(inputs)
                    for future, result in zip(futures, results):
                        if not future.done():
                            future.set_result(result)
                except Exception as e:
                    for future in futures:
                        if not future.done():
                            future.set_exception(e)
    
    def start(self):
        """启动调度器"""
        asyncio.create_task(self._batch_processor())
    
    def stop(self):
        """停止调度器"""
        self._running = False

这段代码的核心逻辑就三步：

1. submit：把请求扔进队列，返回一个future
2. _batch_processor：后台循环，凑够一批就推理
3. 超时控制：用deadline控制最长等待时间

4.4 动态batching的视觉化理解

光看代码可能还不够直观。我画了张流程图，帮你理清整个调度过程：

从这张图你能看到，动态batching本质上是个时间-空间权衡。等得越久，batch越大，吞吐越高，但延迟也上去了。怎么平衡？我一般用这个经验公式：

最优等待时间 ≈ 模型推理时间 × 0.3

比如模型推理要100ms，我就设30ms的等待窗口。这样既不会让用户等太久，又能把batch size提上去。

好了，动态batching的核心思路和实现就这些。你回去可以拿我的代码跑一跑，把max_batch和max_wait调一调，看看吞吐和延迟的变化曲线。实践出真知嘛。