连续批处理内部原理

静态批处理的效率问题

在传统LLM推理中，请求被分成固定大小的batch。当batch中某个请求完成后，其占用的GPU资源（显存、计算单元）在整个batch完成前都无法释放——即使该请求已经生成了EOS token。

这在大语言模型服务中造成了严重的资源浪费。用户请求的长度差异巨大——有的只需生成10个token，有的需要生成1000个token。静态批处理强迫"快请求等待慢请求"，GPU计算能力被大量闲置。

连续批处理的核心思想

连续批处理（Continuous Batching，也称动态批处理）的核心创新是：不再等待整个batch完成，而是在每个生成步骤后动态重组batch。

当一个请求完成时，立即用新请求填充其位置，始终保持GPU满载。

与静态批处理对比

静态批处理（batch_size=4，完成时间120ms）:
时刻0: [Req1, Req2, Req3, Req4] → 开始生成
时刻1: Req1完成（用时20ms），但仍在等待其他请求
时刻2: Req2完成（用时50ms），仍在等待
时刻3: Req3完成（用时80ms），仍在等待
时刻4: Req4完成（用时120ms），batch完成
→ 实际吞吐：4个请求 / 120ms

连续批处理:
时刻0: [Req1, Req2, Req3, Req4] → 开始生成
时刻1: Req1完成 → 立即加入Req5 → [Req5, Req2, Req3, Req4]
时刻2: Req2完成 → 立即加入Req6 → [Req5, Req6, Req3, Req4]
...
→ 实际吞吐：~10个请求 / 120ms（约2.5倍提升！）

vLLM中的连续批处理实现

请求状态机

每个请求在vLLM中经历以下状态：

class RequestState(Enum):
    WAITING = 0    # 等待被调度
    RUNNING = 1     # 正在生成
    PREEMPTED = 2   # 被抢占（显存不足）
    FINISHED = 3    # 完成

class Request:
    def __init__(self, prompt, max_tokens):
        self.state = RequestState.WAITING
        self.prompt = prompt
        self.output_tokens = []
        self.max_tokens = max_tokens
        self.current_step = 0

调度循环

vLLM的调度器在每一步动态决定哪些请求参与计算：

class ContinuousBatchingScheduler:
    def __init__(self, max_batch_size, max_seq_len):
        self.waiting_queue = deque()
        self.running_set = set()
        self.max_batch_size = max_bach_size
        self.max_seq_len = max_seq_len
    
    def schedule(self):
        """每一步的调度决策"""
        # 1. 检查当前运行的请求
        to_remove = []
        for req in self.running_set:
            if self.is_finished(req):
                req.state = RequestState.FINISHED
                to_remove.append(req)
            elif req.current_step >= self.max_seq_len:
                # 达到最大长度，强制完成
                req.state = RequestState.FINISHED
                to_remove.append(req)
        
        # 移除完成的请求
        for req in to_remove:
            self.running_set.remove(req)
        
        # 2. 用新请求填充空位
        while (len(self.running_set) < self.max_batch_size and 
               len(self.waiting_queue) > 0):
            new_req = self.waiting_queue.popleft()
            new_req.state = RequestState.RUNNING
            self.running_set.add(new_req)
        
        # 3. 处理显存不足的情况（抢占）
        if not self.has_enough_memory():
            # 抢占优先级最低的请求
            preempted = self.preempt_lowest_priority()
            self.waiting_queue.appendleft(preempted)
        
        return list(self.running_set)

KV Cache的动态管理

连续批处理的核心挑战是KV Cache的管理——不同请求的生成进度不同，KV Cache的长度也不同。

vLLM使用PagedAttention（之前文章介绍过）解决这个问题：

class PagedKVCacheManager:
    def __init__(self, n_blocks, block_size):
        self.n_blocks = n_blocks
        self.block_size = block_size
        self.free_blocks = list(range(n_blocks))
        self.request_blocks = {}  # req_id -> [block_indices]
    
    def allocate_for_request(self, req_id, prompt_len):
        """为请求分配KV Cache块"""
        n_blocks_needed = (prompt_len + self.block_size - 1) // self.block_size
        
        if len(self.free_blocks) < n_blocks_needed:
            # 需要驱逐一些块的请求
            self.evict_blocks(n_blocks_needed - len(self.free_blocks))
        
        blocks = []
        for _ in range(n_blocks_needed):
            blocks.append(self.free_blocks.pop())
        
        self.request_blocks[req_id] = blocks
    
    def append_block(self, req_id):
        """为请求追加一个新的KV Cache块"""
        if not self.free_blocks:
            self.evict_blocks(1)
        
        new_block = self.free_blocks.pop()
        self.request_blocks[req_id].append(new_block)
        return new_block
    
    def free_request(self, req_id):
        """释放请求的KV Cache块"""
        blocks = self.request_blocks.pop(req_id, [])
        self.free_blocks.extend(blocks)

优先级调度

基于优先级的抢占

不是所有请求都同等重要。vLLM支持基于优先级的调度：

class PriorityScheduler(ContinuousBatchingScheduler):
    def preempt_lowest_priority(self):
        """抢占优先级最低的正在运行请求"""
        # 按优先级排序（高优先级数值小）
        sorted_running = sorted(
            self.running_set,
            key=lambda r: (r.priority, r.arrival_time)
        )
        
        # 抢占优先级最低（数值最大）的请求
        preempted = sorted_running[-1]
        preempted.state = RequestState.PREEMPTED
        
        # 保存其已生成的内容
        self.save_generation_state(preempted)
        
        self.running_set.remove(preempted)
        return preempted
    
    def schedule(self):
        # 抢占后，被抢占的请求重新进入等待队列
        # 其优先级会提升（避免饥饿）
        for req in self.waiting_queue:
            if req.state == RequestState.PREEMPTED:
                req.priority = max(0, req.priority - 1)
        
        return super().schedule()

延迟保证

对于延迟敏感的应用（如聊天机器人），需要保证请求在指定时间内开始处理：

class LatencyGuaranteeScheduler(ContinuousBatchingScheduler):
    def __init__(self, max_latency=5.0):
        super().__init__()
        self.max_latency = max_latency  # 最大等待时间（秒）
    
    def add_request(self, req):
        req.arrival_time = time.time()
        self.waiting_queue.append(req)
    
    def schedule(self):
        current_time = time.time()
        
        # 检查是否有请求等待时间过长
        for req in list(self.waiting_queue):
            wait_time = current_time - req.arrival_time
            if wait_time > self.max_latency:
                # 强制调度（可能超过max_batch_size）
                req.priority = 0  # 最高优先级
                self.preempt_for_latency(req)
        
        return super().schedule()
    
    def preempt_for_latency(self, urgent_req):
        """为紧急请求抢占资源"""
        while len(self.running_set) >= self.max_batch_size:
            preempted = self.preempt_lowest_priority()
            self.waiting_queue.appendleft(preempted)
        
        urgent_req.state = RequestState.RUNNING
        self.running_set.add(urgent_req)
        self.waiting_queue.remove(urgent_req)

与推测解码的结合

连续批处理和推测解码可以结合使用，但需要注意：

class SpeculativeContinuousBatching:
    def schedule_with_speculative(self):
        """结合推测解码的连续批处理"""
        running = self.scheduler.schedule()
        
        # 对运行中的请求应用推测解码
        results = []
        for req in running:
            # 推测解码：Draft模型快速生成候选
            draft_tokens = self.draft_model.generate(
                req, n_draft=self.n_draft
            )
            
            # 验证：大模型并行验证
            accepted = self.target_model.verify(
                req, draft_tokens
            )
            
            req.output_tokens.extend(accepted)
            req.current_step += len(accepted)
            
            if self.is_finished(req):
                results.append(req)
        
        # 完成的请求立即释放，加入新请求
        for req in results:
            self.scheduler.running_set.remove(req)
            self.add_new_requests()
        
        return results

性能分析

吞吐量提升

在混合长度请求场景下，连续批处理相比静态批处理的吞吐量提升：

场景	静态批处理吞吐	连续批处理吞吐	提升
均匀长度（相同）	100 req/s	100 req/s	0%
混合长度（10-1000 tokens）	40 req/s	95 req/s	137%
高度混合（1-4000 tokens）	15 req/s	85 req/s	466%

GPU利用率

连续批处理保持GPU利用率在80-95%，而静态批处理在混合长度场景下可能降至30-50%。

2026年前沿方向

异步连续批处理

将调度与计算完全解耦——调度器在单独的CPU线程中运行，GPU只负责计算。这进一步减少了调度开销。

跨节点连续批处理

在分布式推理中，不同节点上的请求可以动态调度以实现全局负载均衡。

与MoE的结合

在MoE模型中，连续批处理需要考虑专家负载均衡——避免所有请求都路由到少数专家。

实践建议

合适的max_batch_size：太大导致显存不足，太小导致GPU利用率低。根据显存和模型大小调整
优先级设置：为不同应用设置合理的优先级，避免低优先级请求饥饿
KV Cache容量监控：持续监控KV Cache的使用率，动态调整max_batch_size
与量化结合：INT8/INT4量化可以减少KV Cache占用，允许更大的有效batch size

结语

连续批处理是LLM推理系统从"能用"到"高效"的关键技术之一。它让GPU始终保持高利用率，大幅提升了服务吞吐量。vLLM、TensorRT-LLM等主流推理引擎都已经采用了这一技术。随着LLM应用场景的多样化，连续批处理的重要性只会越来越高。

加入讨论

这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞，认知差异创造无限可能。

静态批处理的效率问题#

连续批处理的核心思想#

与静态批处理对比#

vLLM中的连续批处理实现#

请求状态机#

调度循环#

KV Cache的动态管理#

优先级调度#

基于优先级的抢占#

延迟保证#

与推测解码的结合#

性能分析#

吞吐量提升#

GPU利用率#

2026年前沿方向#

异步连续批处理#

跨节点连续批处理#

与MoE的结合#

实践建议#

结语#

加入讨论#

📌 相关推荐