企业官网建设流程全解析

系列导读：前六篇把模型转换和量化调试全部打通。从这篇开始进入性能优化主题。RK3588S 的 NPU 有三个独立核心，但默认情况下单个推理任务只用一个核心——本篇教你用多线程异步推理充分利用全部算力，实现真正意义上的多路并发，吞吐量提升 2-3 倍。

一、为什么默认只用了三分之一算力

先回顾第一篇讲过的 NPU 三核结构：

RK3588S NPU ┌─────────────────────────────────┐ │ Core 0 (2T)│Core 1 (2T)│Core 2 (2T)│ ├─────────────────────────────────┤ │ 共享 L2 Cache + DMA │ └─────────────────────────────────┘

问题的根源：调用rknn_init()时，如果不指定 core mask，SDK 默认只激活 Core 0。单个推理任务跑完整个模型，Core 1 和 Core 2 全程空闲。

验证这一点：

# 在板子上跑推理程序的同时，另开终端查看 NPU 利用率 cat /sys/kernel/debug/rknpu/load # 输出示例： # NPU load: Core0: 87% Core1: 0% Core2: 0% # 三分之二的算力白白浪费

要充分利用 6 TOPS，有两种策略，适合不同场景：

二、策略一：单模型三核并发

2.1 设置 core mask

// single_model_3core.cpp #include "rknn_api.h" #include <stdio.h> int main() { // 加载模型 int model_size = 0; unsigned char* model_data = load_model("model.rknn", &model_size); rknn_context ctx; // 初始化时不指定核心（后面单独设置） int ret = rknn_init(&ctx, model_data, model_size, 0, NULL); free(model_data); // ── 关键：设置三核并发 ── ret = rknn_set_core_mask(ctx, RKNN_NPU_CORE_0_1_2); if (ret < 0) { fprintf(stderr, "设置 core mask 失败: %d\n", ret); return -1; } // 推理（和之前完全一样，SDK 内部自动分配三核） rknn_input inputs[1] = {0}; inputs[0].index = 0; inputs[0].type = RKNN_TENSOR_UINT8; inputs[0].size = 224 * 224 * 3; inputs[0].fmt = RKNN_TENSOR_NHWC; inputs[0].buf = img_data; inputs[0].pass_through = 0; rknn_inputs_set(ctx, 1, inputs); rknn_run(ctx, NULL); // SDK 内部自动三核调度 rknn_output outputs[1] = {0}; outputs[0].want_float = 1; rknn_outputs_get(ctx, 1, outputs, NULL); // ... 处理结果 ... rknn_outputs_release(ctx, 1, outputs); rknn_destroy(ctx); return 0; }

2.2 单模型三核并发的性能特征

单核推理： Core0: [===========推理============] Core1: [空闲] Core2: [空闲] 耗时：T ms 三核并发： Core0: [===推理===] Core1: [===推理===] Core2: [===推理===] 耗时：约 0.6×T ms（不是 T/3，因为有调度开销和共享 L2 Cache 竞争）

⚠️重要认知：三核并发不能把延迟压缩到原来的 1/3，原因是三核共享 L2 Cache 和 DMA 带宽，存在竞争开销。实际效果是延迟降低 35-45%，吞吐量提升约 2 倍。对于大模型（如 YOLOv8m、ResNet50）效果更明显，小模型（MobileNetV2）的调度开销占比更高，收益相对小。

三、策略二：多模型实例并发（生产场景首选）

3.1 架构设计

多路视频 AI 分析是 RK3588S 最典型的应用场景。以三路摄像头同时做目标检测为例：

摄像头0 → [推理线程0, Core0] ──┐ 摄像头1 → [推理线程1, Core1] ──┼──→ 结果汇总 → 输出 摄像头2 → [推理线程2, Core2] ──┘ 每个核心独立运行，互不干扰 真正的物理并行，总吞吐 ≈ 单核的3倍

3.2 完整多实例并发实现

// multi_instance_infer.cpp #include "rknn_api.h" #include <pthread.h> #include <queue> #include <mutex> #include <condition_variable> #include <atomic> #include <stdio.h> #include <string.h> #define NUM_CORES 3 #define INPUT_SIZE (224 * 224 * 3) // ── 线程参数结构 ────────────────────────────── struct InferThread { int core_id; rknn_context ctx; unsigned char* input_buf; float* output_buf; int output_size; pthread_t thread; std::atomic<bool> task_ready{false}; std::atomic<bool> result_ready{false}; std::mutex mtx; std::condition_variable cv_task; std::condition_variable cv_result; bool running{true}; }; // ── 推理线程主函数 ──────────────────────────── void* infer_worker(void* arg) { InferThread* t = (InferThread*)arg; printf("[Core%d] 推理线程启动\n", t->core_id); while (t->running) { // 等待任务 std::unique_lock<std::mutex> lock(t->mtx); t->cv_task.wait(lock, [t]{ return t->task_ready.load() || !t->running; }); if (!t->running) break; t->task_ready = false; lock.unlock(); // 设置输入 rknn_input inputs[1] = {0}; inputs[0].index = 0; inputs[0].type = RKNN_TENSOR_UINT8; inputs[0].size = INPUT_SIZE; inputs[0].fmt = RKNN_TENSOR_NHWC; inputs[0].buf = t->input_buf; inputs[0].pass_through = 0; rknn_inputs_set(t->ctx, 1, inputs); // 执行推理 rknn_run(t->ctx, NULL); // 获取输出 rknn_output outputs[1] = {0}; outputs[0].want_float = 1; rknn_outputs_get(t->ctx, 1, outputs, NULL); memcpy(t->output_buf, outputs[0].buf, outputs[0].size < (size_t)(t->output_size * sizeof(float)) ? outputs[0].size : t->output_size * sizeof(float)); rknn_outputs_release(t->ctx, 1, outputs); // 通知结果就绪 t->result_ready = true; t->cv_result.notify_one(); } printf("[Core%d] 推理线程退出\n", t->core_id); return nullptr; } // ── core mask 映射 ──────────────────────────── rknn_core_mask get_core_mask(int core_id) { switch (core_id) { case 0: return RKNN_NPU_CORE_0; case 1: return RKNN_NPU_CORE_1; case 2: return RKNN_NPU_CORE_2; default: return RKNN_NPU_CORE_AUTO; } } int main() { // 加载模型文件（三个实例共用同一份模型文件） int model_size = 0; unsigned char* model_data = load_model("model.rknn", &model_size); InferThread threads[NUM_CORES]; // ── 初始化三个推理实例 ───────────────────── for (int i = 0; i < NUM_CORES; i++) { threads[i].core_id = i; // 每个实例独立初始化上下文 int ret = rknn_init(&threads[i].ctx, model_data, model_size, 0, NULL); if (ret < 0) { fprintf(stderr, "[Core%d] rknn_init 失败: %d\n", i, ret); return -1; } // 绑定核心 rknn_set_core_mask(threads[i].ctx, get_core_mask(i)); // 分配 IO 缓冲区 threads[i].input_buf = new unsigned char[INPUT_SIZE]; threads[i].output_size = 1000; // MobileNetV2 输出 1000 类 threads[i].output_buf = new float[1000]; // 启动推理线程 pthread_create(&threads[i].thread, nullptr, infer_worker, &threads[i]); } free(model_data); // 三个实例都初始化完成后释放原始数据 // ── 模拟三路输入并发推理 ─────────────────── printf("\n开始三路并发推理...\n"); for (int frame = 0; frame < 10; frame++) { // 填充三路输入数据（实际场景从摄像头读取） for (int i = 0; i < NUM_CORES; i++) { memset(threads[i].input_buf, frame * 10 + i, INPUT_SIZE); // 触发推理任务 { std::lock_guard<std::mutex> lock(threads[i].mtx); threads[i].task_ready = true; } threads[i].cv_task.notify_one(); } // 等待三路结果 for (int i = 0; i < NUM_CORES; i++) { std::unique_lock<std::mutex> lock(threads[i].mtx); threads[i].cv_result.wait(lock, [&threads, i]{ return threads[i].result_ready.load(); }); threads[i].result_ready = false; // 处理结果 float* probs = threads[i].output_buf; int top1 = std::max_element(probs, probs + 1000) - probs; printf("[Core%d] Frame%d Top-1: %d (%.4f)\n", i, frame, top1, probs[top1]); } } // ── 清理 ────────────────────────────────── for (int i = 0; i < NUM_CORES; i++) { threads[i].running = false; threads[i].cv_task.notify_one(); pthread_join(threads[i].thread, nullptr); rknn_destroy(threads[i].ctx); delete[] threads[i].input_buf; delete[] threads[i].output_buf; } return 0; }

四、异步推理 API：rknn_run_async

除了多线程方案，RKNN SDK 还提供了原生的异步推理接口，适合 pipeline 场景：

// async_infer.cpp // ── 异步推理回调函数 ────────────────────────── void infer_callback(rknn_context ctx, rknn_recv_object* recv_obj, void* userdata) { // 推理完成时被调用（在独立线程中） int frame_id = *(int*)userdata; rknn_output outputs[1] = {0}; outputs[0].want_float = 1; rknn_outputs_get(ctx, 1, outputs, NULL); float* probs = (float*)outputs[0].buf; int top1 = std::max_element(probs, probs + 1000) - probs; printf("Frame %d 推理完成，Top-1: %d\n", frame_id, top1); rknn_outputs_release(ctx, 1, outputs); free(userdata); } // ── 异步推理主流程 ──────────────────────────── int run_async_infer(rknn_context ctx, unsigned char* img_data, int frame_id) { // 设置输入（与同步推理相同） rknn_input inputs[1] = {0}; inputs[0].index = 0; inputs[0].type = RKNN_TENSOR_UINT8; inputs[0].size = 224 * 224 * 3; inputs[0].fmt = RKNN_TENSOR_NHWC; inputs[0].buf = img_data; inputs[0].pass_through = 0; rknn_inputs_set(ctx, 1, inputs); // 异步提交推理任务 int* user_data = new int(frame_id); rknn_run_async(ctx, infer_callback, user_data); // 函数立即返回，不阻塞 // 推理完成时自动调用 infer_callback return 0; }

同步 vs 异步的选择：

五、性能对比实测（YOLOv8n，1080P输入）

💡选哪个？如果只有一路视频但追求最低延迟，用三核并发。如果有多路视频需要同时处理，用多实例各占一核，总吞吐量最大化。

六、内存优化：多实例共享权重

三个模型实例初始化时，如果每个实例都独立加载权重，会占用三倍内存。RKNN SDK 提供了权重共享机制：

// 初始化第一个实例（加载完整模型） rknn_init(&ctx0, model_data, model_size, 0, NULL); rknn_set_core_mask(ctx0, RKNN_NPU_CORE_0); // 第二、三个实例通过 dup_context 共享权重 // 不需要再次加载 model_data，内存占用大幅减少 rknn_dup_context(&ctx0, &ctx1); rknn_set_core_mask(ctx1, RKNN_NPU_CORE_1); rknn_dup_context(&ctx0, &ctx2); rknn_set_core_mask(ctx2, RKNN_NPU_CORE_2); free(model_data); // 三个实例都准备好后再释放 // 内存占用对比： // 不用 dup_context：3 × 模型大小 // 用 dup_context： 1 × 模型大小（权重共享）

⚠️注意：rknn_dup_context共享的是权重（只读），每个实例的激活缓冲区（中间层特征图）是独立的，推理过程互不干扰。

七、线程绑核：防止 CPU 调度抖动

推理线程应该绑定到大核，防止 Linux 调度器把推理线程迁移到小核导致延迟抖动：

#include <pthread.h> #include <sched.h> void bind_thread_to_core(pthread_t thread, int cpu_core) { cpu_set_t cpuset; CPU_ZERO(&cpuset); CPU_SET(cpu_core, &cpuset); pthread_setaffinity_np(thread, sizeof(cpu_set_t), &cpuset); } // 在推理线程启动后调用 // RK3588S 上 cpu 4-7 通常是 A76 大核（具体看 cpufreq） bind_thread_to_core(threads[0].thread, 4); bind_thread_to_core(threads[1].thread, 5); bind_thread_to_core(threads[2].thread, 6);

八、完整性能调优检查清单

多线程推理上线前检查： NPU 侧： ☐ 每个实例已正确设置 core_mask（0/1/2 分别绑定） ☐ 用 /sys/kernel/debug/rknpu/load 验证三核利用率均衡 ☐ 多实例使用 rknn_dup_context 共享权重，节省内存 CPU 侧： ☐ 推理线程绑定到大核（A76，cpu4-7） ☐ 前处理线程和推理线程分配不同大核，避免竞争 ☐ 后处理（NMS等）可以放小核 内存侧： ☐ 确认无内存泄漏（rknn_outputs_release 每次都调用） ☐ 输入缓冲区用 DMA buffer（见第8篇）进一步减少拷贝 验证： ☐ 三路并发结果与单路结果一致（排除共享状态 bug） ☐ 长时间运行（>1小时）无崩溃、无内存增长 ☐ 用 perf 或 top 确认 CPU 占用率在预期范围内

九、总结与下篇预告

本篇完整覆盖了多线程异步推理的两大策略：

• 单模型三核并发：rknn_set_core_mask(ctx, RKNN_NPU_CORE_0_1_2)，延迟降低 40%
• 多模型多实例并发：三线程各占一核，吞吐量提升 3 倍
• rknn_dup_context权重共享节省内存
• CPU 线程绑核防止调度抖动

下一篇（系列第 8 篇）讲RGA 零拷贝图像加速：把图像前处理从 CPU 完全卸载到 RGA 硬件，1080P resize + 格式转换从 12ms 压缩到 1ms 以内，是整个 Pipeline 优化最关键的一环。

本系列文章列表（持续更新）

• ✅ 第1-6篇：已发布，点击专栏查看
• ✅ 第7篇：多线程异步推理（本文）
• 🔜 第8篇：RGA零拷贝图像加速
• … 共16篇