边缘 AI 推理的批量调度器设计在固定延迟窗口内合并请求增加吞吐量的工程实践一、边缘推理的吞吐-延迟矛盾单请求推理的资源浪费在边缘侧部署目标检测或图像分类模型时一个常见的使用模式是摄像头以 30fps 持续送入画面推理服务逐帧处理。在单帧推理模式下GPU/NPU 的利用率经常不足 30%——模型加载、内存分配、上下文切换等固定开销在每帧中重复发生而实际计算矩阵乘/卷积的占比并不高。问题的本质是边缘侧请求到达模式呈周期性帧间隔但请求之间存在未被利用的计算能力余量。批量推理通过等待一小段时间、将多帧合并为一批、一次推理处理多帧来摊薄固定开销。但批量推理又引入了新矛盾单帧的调度延迟增加了。本文设计的批量调度器需要在吞吐提升和延迟恶化之间找到可控的平衡点。二、固定延迟窗口的批量调度模型时间驱动的生产者-消费者管道批量调度器的核心策略是维护一个请求队列设置固定的收集窗口例如 10ms窗口结束时将收集到的所有请求合并为一个 batch送入推理引擎。flowchart TD subgraph Input[请求源摄像帧 / API 调用] R1[帧 1] -- Q R2[帧 2] -- Q R3[帧 3] -- Q RN[帧 N] -- Q end Q[请求队列br/FIFO] --|窗口到达| MERGE[请求合并器br/batch_size queue.size()] MERGE --|batched tensor| ENGINE[推理引擎br/NCNN / TFLite / ONNX] ENGINE --|batched output| SPLIT[结果拆分器br/按 request_id 分发] SPLIT -- CB1[回调: 帧1 结果] SPLIT -- CB2[回调: 帧2 结果] SPLIT -- CB3[回调: 帧N 结果] TIMER[收集窗口定时器br/interval 10ms] -.-|超时触发| MERGE当 batch_size1 时等价于单帧推理。batch_size 的实际值由窗口内的请求数量决定极端情况下可能为 0空 batch直接跳过或超过模型支持的最大 batch_size需降级为多次推理。三、C 实现线程安全的批量调度器以下实现使用条件变量管理请求收集和推理触发之间的同步// batch_scheduler.hpp —— 边缘推理批量调度器 // 设计原则 // 1. 请求线程和推理线程解耦 // 2. 固定延迟窗口max_latency_ms限制单帧最大等待时间 // 3. 最大 batch_size 防止内存爆炸 #pragma once #include queue #include vector #include mutex #include condition_variable #include chrono #include functional #include memory #include cstdint // 单个推理请求的数据结构 struct InferenceRequest { uint64_t request_id; // 请求唯一 ID用于结果匹配 std::vectorfloat input_data; // 模型输入数据已预处理 size_t input_size; // 单个输入的大小字节 // 回调函数推理完成时调用传入输出数据 std::functionvoid(const std::vectorfloat) callback; }; // 批量调度器 class BatchScheduler { public: /* * param max_batch_size 模型支持的最大 batch_size * param window_ms 收集窗口时长毫秒 * param inference_fn 推理函数void(const std::vectorInferenceRequest, * std::vectorstd::vectorfloat) */ BatchScheduler(int max_batch_size, int window_ms, std::functionvoid(const std::vectorInferenceRequest, std::vectorstd::vectorfloat) inference_fn) : max_batch_size_(max_batch_size), window_ms_(window_ms), inference_fn_(std::move(inference_fn)), running_(false) {} ~BatchScheduler() { stop(); } // 启动调度线程只启动一次 void start() { running_ true; scheduler_thread_ std::thread(BatchScheduler::scheduler_loop, this); } // 停止调度线程处理完当前批次后退出 void stop() { { std::lock_guardstd::mutex lock(mutex_); running_ false; } cv_.notify_one(); // 唤醒调度线程以检查 running_ 标志 if (scheduler_thread_.joinable()) { scheduler_thread_.join(); } } /* * 提交推理请求调用方线程 * * return 0成功, -1调度器已停止 */ int enqueue(InferenceRequest req) { std::lock_guardstd::mutex lock(mutex_); if (!running_) { return -1; // 调度器已停止拒绝新请求 } request_queue_.push(std::move(req)); cv_.notify_one(); // 通知调度线程有新请求到达 return 0; } private: // 调度主循环运行在独立线程中 void scheduler_loop() { while (true) { std::vectorInferenceRequest batch; batch.reserve(max_batch_size_); { std::unique_lockstd::mutex lock(mutex_); /* * 等待第一个请求到达或收到停止信号 * 使用 cv_.wait_for 而非 wait_until * 因为窗口从第一个请求到达时开始计时 */ if (request_queue_.empty()) { cv_.wait(lock, [this]() { return !running_ || !request_queue_.empty(); }); } if (!running_ request_queue_.empty()) { break; // 正常退出 } // 等待收集窗口期满或达到最大 batch_size auto deadline std::chrono::steady_clock::now() std::chrono::milliseconds(window_ms_); while (batch.size() (size_t)max_batch_size_) { if (cv_.wait_until(lock, deadline) std::cv_status::timeout) { break; // 窗口期满立即处理当前 batch } // 有新请求到达或 super-batch 触发 if (request_queue_.empty()) continue; batch.push_back(std::move(request_queue_.front())); request_queue_.pop(); // 达到最大 batch_size立即处理 if (batch.size() (size_t)max_batch_size_) break; } // 窗口期满但 batch 仍为空处理剩余请求 while (!request_queue_.empty() batch.size() (size_t)max_batch_size_) { batch.push_back(std::move(request_queue_.front())); request_queue_.pop(); } } // 释放锁后执行推理避免阻塞请求提交 if (!batch.empty()) { std::vectorstd::vectorfloat outputs; try { inference_fn_(batch, outputs); } catch (const std::exception e) { // 推理异常所有请求返回空结果回调会收到空 vector outputs.assign(batch.size(), std::vectorfloat()); } // 逐个回调传递对应结果 for (size_t i 0; i batch.size(); i) { if (batch[i].callback) { batch[i].callback(outputs[i]); } } } } } int max_batch_size_; int window_ms_; std::functionvoid(const std::vectorInferenceRequest, std::vectorstd::vectorfloat) inference_fn_; std::queueInferenceRequest request_queue_; std::mutex mutex_; std::condition_variable cv_; std::thread scheduler_thread_; bool running_; };四、架构权衡窗口参数对吞吐和延迟的定量影响窗口长度window_ms_是最关键的调优参数。以 25fps 视频流帧间隔 40ms为例window_ms平均 batch_size单帧延迟增量吞吐提升适用场景5ms1.12.5ms0%无批量实时控制、避障15ms1.47.5ms18%视频分析、目标检测30ms1.815ms35%离线批量预处理100ms2.950ms65%云端代理转发三项工程约束内存预分配batch 推理需要将多个输入拼接为[N, C, H, W]张量。如果模型输入为 224x224x3约 600KBbatch_size8 时单次输入达 4.8MB。在边缘设备上需预分配内存池避免频繁分配/释放。batch_size 不对齐当队列中请求数小于max_batch_size_时需将输入张量不足的部分补 0。某些算子如 BatchNorm对 padding 敏感需在推理后再移除 padding 输出。延迟抖动窗口机制引入的最大延迟 window_ms_ 推理时间。对于需要确定性响应的工业控制场景应禁用批量调度window_ms_0等价于即时推理。五、总结批量调度器在边缘 AI 推理场景下用可控的延迟增量换取可观的吞吐提升。核心设计决策在于收集窗口的时长设定——它决定了延迟天花板和 batch_size 期望值之间的平衡。实现层面条件变量 超时机制的线程模型在保证线程安全的同时避免了忙等待造成的 CPU 浪费。实际部署时建议通过运行时参数动态调整窗口长度配合监控系统观察 P99 延迟和 GPU/NPU 利用率找到具体场景的最优工作点。