大模型推理的请求调度算法——FCFS、连续批处理与优先级队列的多维对比

亚历克斯神

148人浏览 · 2026-07-15 21:06:49

亚历克斯神 · 2026-07-15 21:06:49 发布

大模型推理的请求调度算法——FCFS、连续批处理与优先级队列的多维对比

一、背景与问题

大模型推理服务的核心瓶颈不在单次推理的延迟，而在并发请求的调度效率。当推理网关同时接收到数十个请求时，如何组织这些请求的执行顺序，直接决定了 GPU 利用率、TPOT（Time Per Output Token）和排队延迟三个核心指标。

我们在部署 Llama-3 70B 模型的生产集群中发现：同样的硬件配置（8×A100，vLLM 框架），仅仅将调度策略从默认的 FCFS（先来先服务）切换为 Continuous Batching（连续批处理），就能将并发吞吐提升 2.3 倍，P99 延迟降低 67%。本文将对比分析三种主流调度算法在生产场景中的表现差异。

二、方案设计

2.1 调度算法的核心矛盾

推理请求调度的特殊之处在于：不同请求的输入/输出长度差异极大（从 50 Token 的问答到 4000 Token 的文档摘要），且输出是逐步生成的（自回归解码）。这导致传统微服务领域的调度策略（如轮询、最少连接）不能直接套用。

flowchart LR
    subgraph "请求到达"
        A1["请求A: 50 Token输入<br/>预计输出 200 Token"]
        A2["请求B: 2000 Token输入<br/>预计输出 800 Token"]
        A3["请求C: 200 Token输入<br/>预计输出 100 Token"]
    end

    subgraph "FCFS 策略"
        B1["A → B → C<br/>总耗时: A等待0+B等待A+C等待A+B"]
    end

    subgraph "Continuous Batching 策略"
        B2["Batch[A]: A执行<br/>Batch[B]: A继续 + B开始<br/>Batch[C]: A完成 + C开始 + B继续<br/>总耗时: 大幅减少"]
    end

    subgraph "优先级队列策略"
        B3["短请求优先<br/>C → A → B<br/>平均等待降低"]
    end

    A1 --> B1
    A2 --> B1
    A3 --> B1
    A1 --> B2
    A2 --> B2
    A3 --> B2
    A1 --> B3
    A2 --> B3
    A3 --> B3

2.2 三种算法的核心特征

算法	核心机制	吞吐量	平均延迟	尾部延迟	实现复杂度
FCFS	严格按到达顺序，一个一个处理	低	高	极高	极低
Continuous Batching	动态组批：完成一个立即插入下一个	高	中	中	中等
优先级队列	按优先级排序，短请求优先	中高	低	中低	较高

三、实战演示

3.1 vLLM 中的调度参数配置

"""
vLLM 推理引擎的调度配置示例。
调度策略通过 SchedulerConfig 控制，max_num_seqs 决定最大并发批处理数。
"""
from vllm import LLM, SamplingParams
from vllm.config import SchedulerConfig
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


def create_inference_engine_with_scheduler(
    model_name: str,
    max_batch_size: int = 256,
    max_num_seqs: int = 64,
    policy: str = "fcfs"  # fcfs 或 priority
):
    """
    创建配置了特定调度策略的推理引擎。

    关键参数说明：
    - max_num_seqs：同时处理的序列数上限，直接影响 Continuous Batching 的并发度
    - max_num_batched_tokens：单次迭代的最大 Token 数，决定批处理效率
    """
    try:
        scheduler_config = SchedulerConfig(
            max_num_batched_tokens=max_batch_size,
            max_num_seqs=max_num_seqs,
            policy=policy
        )

        llm = LLM(
            model=model_name,
            tensor_parallel_size=8,
            gpu_memory_utilization=0.9,
            scheduler_config=scheduler_config
        )
        logger.info(
            "推理引擎初始化完成: model=%s, max_seqs=%d, policy=%s",
            model_name, max_num_seqs, policy
        )
        return llm

    except Exception as e:
        logger.error("推理引擎初始化失败: %s", e)
        raise RuntimeError(f"引擎创建异常: {e}") from e


def benchmark_scheduler_performance(llm: LLM, prompts: list[str], 
                                     output_len: int = 256):
    """
    基准测试：并发发送请求，统计吞吐与延迟。
    """
    sampling_params = SamplingParams(
        temperature=0.0,
        max_tokens=output_len
    )

    start_time = time.time()
    outputs = llm.generate(prompts, sampling_params)
    elapsed = time.time() - start_time

    total_tokens = sum(len(o.outputs[0].token_ids) for o in outputs)
    throughput = total_tokens / elapsed

    latencies = []
    for o in outputs:
        latencies.append(o.metrics.arrival_time)  # 简化：使用首 Token 时间

    return {
        "throughput_tokens_per_sec": throughput,
        "avg_latency_ms": sum(latencies) / len(latencies) * 1000,
        "p99_latency_ms": sorted(latencies)[int(len(latencies) * 0.99)] * 1000,
        "elapsed_sec": elapsed
    }

3.2 自定义优先级调度器

vLLM 原生支持基于请求到达时间的优先级，但在企业场景中，我们需要根据业务优先级（如付费用户优先、短请求优先）进行调度。以下是一个自定义调度策略的实现思路：

/**
 * 自定义推理请求优先级调度器。
 * 
 * 设计原则：短请求优先（SJF近似） + 业务优先级权重，
 * 同时通过老化机制避免长请求被饿死。
 */
public class InferenceRequestScheduler {

    /**
     * 计算请求的综合优先级分数。
     * 分数越低，优先级越高（类似优先队列的默认排序）。
     *
     * @param request 推理请求
     * @param waitingTimeMs 已等待时间（毫秒）
     * @return 优先级分数
     */
    public static double calculatePriority(InferenceRequest request, 
                                            long waitingTimeMs) {
        try {
            // 期望 Token 数量的归一化值（估计）
            int estimatedTokens = request.getEstimatedOutputTokens();
            double tokenFactor = Math.log1p(estimatedTokens);  // 对数平滑

            // 业务优先级（1-5，1为最高）
            int businessPriority = request.getBusinessPriority();
            double priorityFactor = businessPriority * 2.0;

            // 老化因子：每等待 1000ms，优先级提升 1 分
            double agingFactor = waitingTimeMs / 1000.0;

            // 综合分数 = Token因子 + 业务权重 - 老化补偿
            double score = tokenFactor + priorityFactor - agingFactor;

            // 分数裁剪到 [0, +∞)
            return Math.max(0.0, score);

        } catch (Exception e) {
            throw new IllegalArgumentException(
                "优先级计算失败: " + e.getMessage(), e);
        }
    }

    /**
     * 从待处理队列中选择下一批请求。
     * 使用贪心策略：选择优先级最高的 N 个请求。
     * 
     * @param backlog 待处理请求队列
     * @param maxBatchSize 每批最大请求数
     * @param maxTokens 每批最大 Token 数
     * @return 被选中的请求列表
     */
    public List<InferenceRequest> selectBatch(
            PriorityQueue<InferenceRequest> backlog,
            int maxBatchSize,
            int maxTokens) {

        List<InferenceRequest> batch = new ArrayList<>();
        int currentTokens = 0;

        // 临时队列：处理完后再放回未选中的请求
        List<InferenceRequest> temp = new ArrayList<>();

        while (!backlog.isEmpty() && batch.size() < maxBatchSize) {
            InferenceRequest req = backlog.poll();

            if (req == null) {
                break;
            }

            int reqTokens = req.getEstimatedOutputTokens();

            // 检查 Token 预算
            if (currentTokens + reqTokens > maxTokens) {
                temp.add(req);  // 暂存，待下批处理
                continue;
            }

            batch.add(req);
            currentTokens += reqTokens;
        }

        // 将未选中的请求放回队列
        for (InferenceRequest req : temp) {
            backlog.offer(req);
        }

        return batch;
    }
}