rtmp推理处理

admin

2025-04-25 c3c0d2742db73d8a91cdc28218a58e4863086095

 main.cpp

@@ -16,8 +16,8 @@
// 异步初始化线程池（改为异步任务）
void AsyncInitThreadPool(PreProcessFn& ppf, const Json::Value& ipcsNode)
{ 
    std::vector<std::queue<Mat>> queMatVec(ipcsNode["ipc"].size());//视频队列1
    std::vector<std::queue<Mat>> queMatVec2(ipcsNode["ipc"].size());//视频队列2
    std::vector<std::queue<Mat>> queMatVec(ipcsNode["cameras"].size());//视频队列1
    std::vector<std::queue<Mat>> queMatVec2(ipcsNode["cameras"].size());//视频队列2

    std::vector<std::future<void>> getVideoTasks;//视频取流任务队列
    std::vector<std::future<void>> recognitionTasks;//视频处理任务队列
@@ -27,22 +27,31 @@
    {
        ppf.asyncStop = false;

        for (auto i = 0; i < ipcsNode["ipc"].size(); ++i) {
            Json::Value ipcNode = ipcsNode["ipc"][i];
        for (auto i = 0; i < ipcsNode["cameras"].size(); ++i) {
            Json::Value ipcNode = ipcsNode["cameras"][i];
            std::string id = ipcNode["id"].asString();
            std::string ipccode = ipcNode["ipccode"].asString();        
            std::string ipccode = ipcNode["code"].asString();     
            std::string name = ipcNode["name"].asString();       
            std::string video = ipcNode["video"].asString();
            std::string pushflow = ipcNode["pushflow"].asString();
            Json::Value modelsNode = ipcNode["models"];//模型节点
            long skipN = stol(ipcNode["skipN"].asString()); 
            ppf.skipN = skipN;
            std::string push_type = ipcNode["push_type"].asString();
            std::string fps = ipcNode["fps"].asString();
            std::string toRtmp = ipcNode["toRtmp"].asString();
            std::string interval = ipcNode["interval"].asString();
            Json::Value modelArray = ipcNode["model"];//模型节点    

            if (!modelsNode.isNull()) {
                Json::Value modelArray = modelsNode["model"];        

                // 创建并启动异步任务
               
            if (!modelArray.isNull()) {
                 // 创建并启动异步任务
                getVideoTasks.push_back(std::async(std::launch::async, &PreProcessFn::fnGetVideoMatToQueue, &ppf,video, std::ref(queMatVec[i])));
                recognitionTasks.push_back(std::async(std::launch::async, &PreProcessFn::fnImRecognitionPro, &ppf, ipcNode,modelArray, std::ref(queMatVec[i]), std::ref(queMatVec2[i])));
                pushVideoTasks.push_back(std::async(std::launch::async, &PreProcessFn::fnPushVideoInRabbitMQ, &ppf, std::ref(queMatVec2[i]), ipccode));
                //pushVideoTasks.push_back(std::async(std::launch::async, &PreProcessFn::fnPushVideo, &ppf, std::ref(queMatVec2[i]), ipccode));
               if (push_type == "rabbitmq") {
                    pushVideoTasks.push_back(std::async(std::launch::async, &PreProcessFn::fnPushVideoInRabbitMQ, &ppf, std::ref(queMatVec2[i]), ipccode));
               }
               else if (push_type == "stream") {
                   pushVideoTasks.push_back(std::async(std::launch::async, &PreProcessFn::fnPushVideoToUrl, &ppf, std::ref(queMatVec2[i]),toRtmp,fps, ipccode));
               }
            }  

            //初始化时加个延迟
@@ -101,28 +110,41 @@
        
        Json::Value root;
        config_file >> root; 

        
        // 初始化线程池（改为异步任务）
        PreProcessFn ppf;
        ppf.coalCode= root["ipcs"]["coalCode"].asString();//煤矿编号;
        ppf.warningFilePath = root["ipcs"]["warningFilePath"].asString();//录像保存路径
        ppf.redispath = root["ipcs"]["redispath"].asString();
        ppf.redisport = root["ipcs"]["redisport"].asString();
        ppf.redispass = root["ipcs"]["redispass"].asString();
        ppf.mysqlpath = root["ipcs"]["mysqlpath"].asString();
        ppf.mysqlport = root["ipcs"]["mysqlport"].asString();
        ppf.mysqluser = root["ipcs"]["mysqluser"].asString();
        ppf.mysqlpass = root["ipcs"]["mysqlpass"].asString();
        ppf.mysqldatabase = root["ipcs"]["mysqldatabase"].asString();
        ppf.rabbitpath = root["ipcs"]["rabbitpath"].asString();
        ppf.rabbitport = root["ipcs"]["rabbitport"].asString();
        ppf.rabbituser = root["ipcs"]["rabbituser"].asString();
        ppf.rabbitpass = root["ipcs"]["rabbitpass"].asString();
        ppf.deley = stol(root["ipcs"]["deley"].asString());//跳帧 
        ppf.recordtime = stoi(root["ipcs"]["recordtime"].asString());//录像时间
        ppf.RecIntervalime = stoi(root["ipcs"]["RecIntervalime"].asString());//录像间隔时间 
        ppf.coalCode= root["ipcs"]["coal_code"].asString();//煤矿编号;
        ppf.warningFilePath = root["ipcs"]["warning_path"].asString();//录像保存路径
        ppf.redispath = root["redis"]["ip"].asString();
        ppf.redisport = root["redis"]["port"].asString();  
        ppf.redispass = root["redis"]["pwd"].asString();
        ppf.mysqlpath = root["mysql"]["ip"].asString();
        ppf.mysqlport = root["mysql"]["port"].asString();
        ppf.mysqluser = root["mysql"]["user"].asString();
        ppf.mysqlpass = root["mysql"]["pwd"].asString();
        ppf.mysqldatabase = root["mysql"]["dbname"].asString();
        ppf.rabbitpath = root["rabbit"]["ip"].asString();
        ppf.rabbitport = root["rabbit"]["port"].asString();
        ppf.rabbituser = root["rabbit"]["user"].asString();
        ppf.rabbitpass = root["rabbit"]["pwd"].asString();
        //ppf.deley = stol(root["ipcs"]["deley"].asString());//跳帧 
        ppf.recordtime = stoi(root["ipcs"]["record_time"].asString());//录像时间
        //ppf.RecIntervalime = stoi(root["ipcs"]["RecIntervalime"].asString());//录像间隔时间 
//加载模型标签颜色
            for (int i = 0; i < root["label_colors"].size(); i++)
            {
               Json::Value members = root["label_colors"][i];

        AsyncInitThreadPool(ppf, root["ipcs"]); 
               int nsize = members.size();
               std::vector<int> numbers;
               for (int j = 0; j < nsize; j++)
               {
                  int crood = members[j].asInt();
                  numbers.push_back(crood);
               }
               ppf.c_list.push_back(numbers);
            }
        AsyncInitThreadPool(ppf, root); 

    }
    catch(const std::exception& e)