咨询TBB聚合节点搭建方法：如何实现每接收N张图片输出单结果

Hey there, no need to apologize at all—we all start somewhere with TBB! 😊 针对你需要的累计N张图像后输出单个结果的需求，TBB Flow Graph确实没有直接对应的预制节点，但我们可以通过自定义一个带内部状态的节点来实现，下面我会一步步带你完成：

核心思路 #

我们要创建一个自定义节点，它需要：

• 维护一个内部的FIFO缓冲区（比如std::queue或std::vector）来存储输入的图像
• 每次接收新图像时，将其加入缓冲区，检查是否达到N张
• 当缓冲区满N张时，执行计算逻辑，输出结果，然后清空缓冲区（如果是滑动窗口需求可以调整策略，但你提到固定N张，所以清空即可）

1. 自定义聚合节点类 #

我们可以基于TBB的function_node来扩展，它允许我们方便地维护内部状态，比直接实现sender/receiver接口更简单。假设你的图像类型是cv::Mat（如果是其他类型，直接替换即可），计算结果类型自定义为YourResultType，代码示例如下：

#include <tbb/flow_graph.h>
#include <queue>
#include <mutex>
#include <optional>
// 假设使用OpenCV图像类型，可替换为你实际的图像类型
#include <opencv2/opencv.hpp>

using namespace tbb::flow;

class ImageAggregatorNode {
private:
    std::queue<cv::Mat> image_buffer;
    const size_t batch_size; // 固定参数N
    std::mutex buffer_mutex; // 线程安全锁，避免多线程访问缓冲区冲突
    function_node<cv::Mat, std::optional<cv::Mat>> node;

    // 处理输入图像的核心逻辑
    std::optional<cv::Mat> process_image(const cv::Mat& img) {
        std::lock_guard<std::mutex> lock(buffer_mutex);
        image_buffer.push(img);

        // 检查是否达到批量要求
        if (image_buffer.size() >= batch_size) {
            cv::Mat result = compute_batch_result(image_buffer);
            // 清空缓冲区，准备接收下一批
            while (!image_buffer.empty()) {
                image_buffer.pop();
            }
            return std::make_optional(result);
        } else {
            // 未凑够N张，返回空值
            return std::nullopt;
        }
    }

    // 自定义批量计算逻辑：替换成你的实际业务需求
    cv::Mat compute_batch_result(std::queue<cv::Mat>& images) {
        // 示例：计算N张图像的平均像素值
        cv::Mat avg_img = cv::Mat::zeros(images.front().size(), images.front().type());
        while (!images.empty()) {
            avg_img += images.front() / static_cast<double>(batch_size);
            images.pop();
        }
        return avg_img;
    }

public:
    // 构造函数：传入graph对象和批量大小N
    ImageAggregatorNode(graph& g, size_t n) 
        : batch_size(n), 
          node(g, unlimited, [this](const cv::Mat& img) {
              return process_image(img);
          }) {}

    // 暴露输入端口，方便连接上游节点
    receiver<cv::Mat>& input() {
        return node.input();
    }

    // 暴露输出端口，方便连接下游节点
    sender<std::optional<cv::Mat>>& output() {
        return node.output();
    }
};

2. 集成到Flow Graph管线中 #

现在可以把这个自定义节点加入到你的图像流处理流程里，示例代码如下：

int main() {
    tbb::flow::graph g;

    // 模拟上游图像输入节点：实际场景中可以是摄像头读取、文件流等
    function_node<void, cv::Mat> image_source(g, unlimited, []() {
        // 生成随机测试图像，替换为你的实际输入逻辑
        cv::Mat img = cv::Mat::randu(cv::Size(640, 480), CV_8UC3);
        return img;
    });

    // 创建聚合节点，设置批量大小N=5
    const size_t batch_size = 5;
    ImageAggregatorNode aggregator(g, batch_size);

    // 下游结果处理节点：接收聚合后的结果并处理
    function_node<std::optional<cv::Mat>, void> result_processor(g, unlimited, [](const std::optional<cv::Mat>& opt_result) {
        if (opt_result.has_value()) {
            cv::Mat result = opt_result.value();
            // 示例：保存结果图像，替换为你的实际处理逻辑
            cv::imwrite("batch_result.jpg", result);
        }
        // 忽略空结果（未凑够N张时的输出）
    });

    // 连接整个处理管线
    make_edge(image_source, aggregator.input());
    make_edge(aggregator.output(), result_processor.input());

    // 启动graph并等待所有任务完成
    g.wait_for_all();

    return 0;
}

关键注意事项 #

• 线程安全：如果你的节点设置了多线程并发度（比如示例中的unlimited），一定要给缓冲区加互斥锁，避免多线程同时读写导致的数据混乱。
• 策略调整：如果需要滑动窗口聚合（比如每收到一张新图，用最近N张计算），可以把std::queue换成固定大小的循环缓冲区，满了就移除最旧的图像。
• 错误处理：可以在process_image中添加输入图像有效性检查（比如是否为空、尺寸是否符合要求），避免无效数据进入缓冲区。

希望这个方案能帮到你，如果有其他细节需要调整，随时补充！

内容的提问来源于stack exchange，提问作者Jonn Dove