系列导读：前六篇把模型转换和量化调试全部打通。从这篇开始进入性能优化主题。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，有两种策略，适合不同场景：

策略	做法	适合场景	效果
单模型三核并发	一个模型实例同时使用三个核心	大模型、极致低延迟	单次推理延迟降低约40%
多模型多实例并发	三个模型实例各占一个核心	多路视频分析、多任务AI	吞吐量提升约3倍

---

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

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 异步的选择：

场景	推荐方式
单路视频，简单逻辑	同步 rknn_run
多路视频，需要并发	多线程 + 同步 rknn_run（每线程一个 ctx）
Pipeline 流水线（前处理→推理→后处理重叠）	rknn_run_async
需要最低延迟	同步 + 三核 RKNN_NPU_CORE_0_1_2

---

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

配置	单次延迟	吞吐量（FPS）
单实例，单核（默认）	约 8ms	约 125 FPS
单实例，三核并发	约 5ms	约 200 FPS
三实例，各占一核	约 8ms/路	约 375 FPS（三路合计）
三实例 + 异步推理	约 7ms/路	约 430 FPS（三路合计，流水线重叠）

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

---

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

三个模型实例初始化时，如果每个实例都独立加载权重，会占用三倍内存。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篇