模型压力测试与并发推理评估

可视化演示

压力测试基础

想象一位新上任的餐厅店长，第一天上班就遇上午市高峰。聪明的店长不会直接打开大门让几百号人涌进来——那只会让厨房瞬间瘫痪。他会先约三五好友来试菜，再逐步增加预约桌数，直到发现"再加一桌就要翻车"的临界点。这种步步试探、层层加码的做法，正是压力测试中"阶梯加压"的思路原型。

AI模型推理服务的压力测试，遵循的是同一套逻辑。我们不能一上来就灌入海量请求，那只会得到一堆毫无意义的错误日志。正确做法是像爬楼梯一样，一级一级地增加并发请求数，每级停留一段时间观察系统表现，直到定位到性能拐点——也就是吞吐量不再上升、延迟开始恶化的那个临界并发数。

在具体操作上，压力测试通常分三步走。第一是慢启动预热：模型服务刚启动时，GPU的算子库、显存缓存、CUDA上下文都还没就绪，直接上高并发会得到虚低的数据。先以低并发跑几分钟，等"灶台烧热了"再开始加压。第二是阶梯加压：设置从低到高的多个并发档位，每档保持固定时长（如60秒）并记录指标，找到性能拐点。第三是极限破坏测试：在阶梯加压基础上猛然将并发推到预期上限的数倍，验证系统崩溃时是优雅降级还是直接雪崩——就像故意让厨房接待十倍于容量的客人，看它是先出不了菜还是干脆熄火。

这个公式告诉你每一档压测该设置多少并发：从起始值开始，均匀地往上加，直到目标极限值。

关键性能指标

厨房里一定有一块"运营仪表盘"：今天出了多少道菜、平均上菜时间多久、每口锅的火候利用率如何。没有这块仪表盘，店长就是个瞎子；有了它，哪个环节出问题一目了然。AI服务的性能评估同样需要这样一块仪表盘。

在模型压力测试中，仪表盘上最重要的四个指标是吞吐量、延迟、资源利用率和错误率。它们从不同维度描述了推理服务的健康状况，缺一不可。

吞吐量（QPS）指单位时间内成功处理的请求数，它衡量的是厨房的"出菜速度"。延迟不能只看平均值——少数慢请求会拉高均值，掩盖绝大多数用户的真实体验，所以要看P50、P95、P99分位数。P95意味着95%的请求都在这条线以内，比平均值更能反映"大多数人的感受"。资源利用率包括GPU利用率、显存占用率、CPU使用率，它告诉你瓶颈到底在哪个环节——是锅不够用，还是切菜的人不够。错误率则是压测中不可忽视的警报器，过载时错误率攀升往往比延迟恶化更致命。

吞吐量就是单位时间内成功的请求数。把"成功"二字圈出来很重要——报错的不算数。

并发推理机制

到了饭点，餐厅里坐满了客人。经验丰富的大厨不会一桌一桌地炒菜，而是把相同做法的菜品合到一锅里一起炒——这叫"一锅多炒"。但锅就那么大，一锅炒太多反而会糊。AI的GPU推理，正是这个道理。

GPU天生适合并行计算。当多个推理请求同时到达时，它们可以被"打包"成一个批次（batch）在GPU上并行处理。但批处理的效率不是越高越好，它受到显存、算力、队列的三重约束，这三者形成了一场微妙的"三方博弈"。

首先是GPU显存占用。一批请求需要同时驻留在显存中，批次越大显存占用越高，超出容量就会直接OOM（内存溢出），就像一锅塞太多食材把锅盖掀翻。其次是批处理效率。理论上批次越大、单请求分摊的计算成本越低，但受限于GPU的线程调度和显存带宽，批次增长到一定程度后加速效果会急剧递减，就像锅太大导致受热不均，反而炒不熟。最后是队列等待时间。不是所有请求都能立刻被批处理，它们需要排队；队列太长，延迟就会飙升——就像客人等位太久，即使菜做得再快，体验也已经崩了。

加速比 = 单请求延迟 ÷ 批处理总延迟。它衡量"一锅多炒"到底省了多少时间。

测试方案设计

一场成功的厨房压力测试，绝不能靠"碰运气"。你需要一份剧本：几点到几点接待多少桌客人、点什么菜、后厨记录什么数据。没有剧本的演习，只是闹剧。模型压力测试的"剧本"就是测试方案设计。

一个完整的压测方案，必须回答四个问题：测什么场景、用什么数据、怎么加压、记什么指标。这四件事想清楚了，压测才不是"拿枪乱扫"。

第一个问题是场景设计。线上流量不是均匀的：有早晚高峰、有突发流量、有热点事件引爆的峰值。因此要设计多种场景：均匀低负载的日常场景、阶梯递增的加压场景、瞬间冲击的突发场景。第二个问题是数据构造。压测请求的数据分布要与线上真实流量接近——如果线上是中文短文本，压测却用英文长文本，结果毫无参考价值。就像试菜要用真实的招牌菜，不能拿清水煮白菜来测试顾客的耐心。第三个问题是加压方式。阶梯加压适合找拐点，极限破坏适合验底线，两者结合才能覆盖"从正常到崩溃"的全过程。第四个问题是监控采集。除了QPS和延迟这类聚合指标，还要记录GPU利用率、显存曲线、队列长度等细粒度时序数据，否则出了问题无从定位——就像摄像头只拍了门口，没拍厨房，根本不知道是哪口锅先冒了烟。

这是Little定律的反向使用：想支撑目标QPS，系统里至少要能容纳多少并发请求。

容量规划

经过一场成功的压力测试，你终于知道厨房的极限是同时接待50桌客人。下一个问题是：我的餐厅要开多大？按平均20桌来装修，还是按峰值50桌来装修？——这就是容量规划。

容量规划的目标，是根据压力测试的结果，决定部署多少GPU实例来承载线上流量。它的本质是"用测试数据说话"，而不是拍脑袋定数量。规划做得好，服务上线后既不会浪费资源，也不会在流量高峰时猝不及防。

容量规划的核心步骤是：先分析压测数据，找出单实例的最优QPS和对应延迟；再预测线上峰值流量（通常参考历史监控数据）；最后加上安全余量，因为线上流量永远比预测更调皮。SLA（服务等级协议）在这里起到约束作用——如果你的SLA要求P99延迟在200ms以内，那么压测数据中只有满足该延迟的QPS才能作为容量依据。冗余设计是最后一环：至少预留30%的余量，以应对流量突刺和单机故障。这就像餐厅装修时，按"平时30桌+周末45桌+突发50桌"来预留空间，而不是只按"平均水平"来设计。

需要几台机器 = 峰值QPS ÷（单台机器的QPS × 安全系数），然后向上取整。

弹性伸缩配置

聪明的餐厅老板不会全年都养着应付50桌客人的厨师团队。周五晚上客人多就临时加聘帮厨，周一到周三客人少就让帮厨休息。这种随客流变化调整人手的策略，就是弹性伸缩。

在AI推理服务中，流量同样有潮汐效应：白天高、深夜低，大促时暴涨、平时平稳。弹性伸缩就是让GPU实例数量跟着流量曲线自动"呼吸"——流量涨时自动扩容，流量退时自动缩容，既保证服务质量，又不浪费算力资源。

实现弹性伸缩首先要定义触发条件。最常用的指标有：QPS超过阈值、P99延迟超过SLA目标、GPU利用率超过80%、队列长度超过预设上限。其次是扩缩容策略：扩容要快，缩容要稳。扩容可以激进些，因为缺算力会直接导致用户体验下降；缩容则要避免频繁抖动，防止"刚缩完又被迫扩容"的震荡。冷启动问题是弹性伸缩的最大挑战：新实例从启动到加载完模型、完成预热，可能需要几十秒甚至几分钟。如果流量已经打过来才开始扩容，用户早就等得不耐烦了。所以生产上常采用"提前扩容"或"保留最小缓冲实例"的策略。最后是成本控制：弹性伸缩的终极目标是在满足SLA的前提下最大化资源利用率，把每一分GPU成本都花在刀刃上。

平均实例数 = 统计周期内每个时刻实例数的平均值。它决定了你的长期成本。

AI 应用场景

Python 演示

"""模拟推理服务压力测试与容量估算"""import math  class InferenceService:    ""“模拟带显存约束和过载退化的推理服务”""     def init(self, gpu_mem=16.0, mem_per_req=0.2, base_latency=0.05):        self.gpu_mem = gpu_mem        self.mem_per_req = mem_per_req        self.base_latency = base_latency     def max_concurrency(self):        ""“根据显存算出理论最大并发数”""        return int(self.gpu_mem / self.mem_per_req)     def latency(self, concurrency):        ""“模拟延迟：正常区批处理加速，过载区急剧上升”""        cap = self.max_concurrency()        if concurrency <= cap:            speedup = math.log2(concurrency + 1)            return max(self.base_latency / speedup, 0.005)        overload = concurrency / cap        return self.base_latency  (1 + 3  (overload - 1) ** 2)     def error_rate(self, concurrency):        ""“模拟错误率：超过容量1.2倍后开始出现错误”""        cap = self.max_concurrency()        if concurrency <= cap * 1.2:            return 0.0        return min((concurrency - cap * 1.2) / cap, 0.5)  def run_stress_test(service, max_c=200, step=10):    ""“阶梯加压：从低并发逐步增加，记录性能指标”""    rows = []    for c in range(step, max_c + 1, step):        lat = service.latency(c)        err = service.error_rate(c)        qps = c / lat * (1 - err)        mem = min(c * service.mem_per_req, service.gpu_mem)        rows.append((c, qps, lat  1000, err  100, mem))    return rows  def suggest_instances(service, target_qps, safety=0.7):    ""“根据压测结果估算满足目标QPS所需的实例数”""    rows = run_stress_test(service)    peak = max(rows, key=lambda r: r[1])    per_instance = peak[1] * safety    return max(1, math.ceil(target_qps / per_instance)), peak[0]  def main():    svc = InferenceService()    rows = run_stress_test(svc)    print(“=== 推理服务压力测试模拟 ===”)    print(f“{'并发数':>6} {'有效QPS':>10} {'延迟(ms)':>10} {'错误率%':>8} {'显存(GB)':>8}”)    for c, qps, lat, err, mem in rows:        print(f"{c:>6} {qps:>10.1f} {lat:>10.1f} {err:>8.1f} {mem:>8.1f}")     instances, best_c = suggest_instances(svc, target_qps=1000)    print(f“\n最佳并发点：{best_c}, 单实例有效QPS峰值约：”          f"{max(rows, key=lambda r: r[1])[1]:.1f}")    print(f“目标QPS=1000, 建议部署实例数：{instances}”)  if name == "main":    main()

常见误区

面试题