4. 吞吐量优化
任务目标
本案例通过实现一个动态批处理推理服务，对比不同批大小（batch size）对模型推理吞吐量的影响。核心目标是：在给定固定测试数据集和推理函数的前提下，量化批处理带来的吞吐量提升，并观察批大小与吞吐量之间的变化趋势。
环境与依赖
需要 Python 3.8+，仅使用标准库 time 和 statistics，无需安装第三方框架。为模拟真实推理负载，使用一个计算密集型函数代替实际模型（如矩阵乘法），保证 CPU 时间可测量且结果可复现。
关键完整代码
import time
import statistics
from concurrent.futures import ThreadPoolExecutor
# ---------- 模拟模型推理（计算密集型，代替真实模型） ----------
def mock_inference(sample):
"""模拟单样本推理：执行固定次数的浮点运算"""
x = float(sample)
for _ in range(2000):
x = (x * 1.0001 + 0.3) % 100.0
return x
# ---------- 批处理推理 ----------
def batch_inference(batch):
"""对一批样本执行推理，返回结果列表"""
return [mock_inference(s) for s in batch]
# ---------- 吞吐量测试 ----------
def measure_throughput(dataset, batch_size, num_workers=4):
"""
按指定批大小处理整个数据集，返回吞吐量（样本/秒）
dataset: 输入样本列表
batch_size: 每批样本数
num_workers: 并行工作线程数（模拟多请求并发）
"""
# 将数据集切分为固定大小的批次
batches = [dataset[i:i+batch_size] for i in range(0, len(dataset), batch_size)]
start = time.perf_counter()
with ThreadPoolExecutor(max_workers=num_workers) as executor:
# 并发提交所有批次（模拟服务端并发处理多个请求）
futures = [executor.submit(batch_inference, b) for b in batches]
results = [f.result() for f in futures]
elapsed = time.perf_counter() - start
total_samples = sum(len(b) for b in batches)
return total_samples / elapsed, results
# ---------- 主程序：对比不同批大小 ----------
if __name__ == "__main__":
# 固定测试数据集：1000个样本
dataset = list(range(1000))
# 测试不同批大小
batch_sizes = [1, 4, 16, 64, 256]
print(f"{'Batch Size':>10} | {'Throughput (samples/s)':>22} | {'Speedup vs bs=1':>16}")
print("-" * 55)
baseline = None
for bs in batch_sizes:
# 每个配置运行3次取中位数，减少噪声
measurements = []
for _ in range(3):
tput, _ = measure_throughput(dataset, bs)
measurements.append(tput)
median_tput = statistics.median(measurements)
if baseline is None:
baseline = median_tput
speedup = 1.0
else:
speedup = median_tput / baseline
print(f"{bs:>10} | {median_tput:>22.2f} | {speedup:>16.2f}x")
运行方式
python throughput_benchmark.py验证方法与判定标准
程序运行后输出一个对比表格。判定优化有效的标准如下：
- 功能正确性：程序无异常退出，且每个批大小配置均输出吞吐量数值。
- 优化效果：批大小从1增大到4或16时，吞吐量应显著提升（通常提升2倍以上）。由于
mock_inference是纯计算函数，批处理减少了线程调度和函数调用开销，因此批大小越大，吞吐量越高。 - 趋势合理性：吞吐量随批大小增大呈上升趋势，但增速逐渐放缓（边际收益递减）。当批大小达到256时，吞吐量应高于批大小为1时的数值。
若观察到批大小增大后吞吐量反而下降，说明批大小已超过CPU并行度上限（本机核心数×每核超线程数），此时应缩小测试范围（如将最大批大小设为64）。
预期输出示例
Batch Size | Throughput (samples/s) | Speedup vs bs=1
-------------------------------------------------------
1 | 125.30 | 1.00x
4 | 312.85 | 2.50x
16 | 498.20 | 3.98x
64 | 601.75 | 4.80x
256 | 623.10 | 4.97x以上数值仅为逻辑一致的示例，实际数值取决于CPU性能与当前负载，但趋势应保持一致。
常见问题
- 吞吐量波动较大：可增加重复次数（如5次）或使用更大数据集（如5000样本）来平滑噪声。
- 批大小增大后吞吐量下降：说明批大小已超过硬件并行能力，应减小最大批大小。
- 线程数设置：
num_workers=4模拟4个并发请求。若本机核心数较少，可调低该值避免上下文切换开销。