devbox
我家自动化
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

目标检测多线程框架(1)_图像帧 队列 多线程 目标检测

作者：我家自动化 | 2024-03-24 05:54:33

赞

踩

图像帧 队列 多线程 目标检测

目标检测用到的异步多线程处理(1)-之异步处理   完整工程下载

采用C11以上语法支持

这里为了好说明问题就不用ffmpeg ,保证代码足够简单易懂以下还是用opencv的VideoCapture描述为主.

通常在目标检测分析处理中大致思路是下面这样的:

   如下:

  1. //视频模拟camera
  2. void testCPUvideo(const char* name)
  3. {
  4.     cv::VideoCapture vcap;
  5.     cv::Mat fps_curr;
  6.     vcap.open(name);      
  7.     vcap.set(CAP_PROP_POS_FRAMES, 1);
  8.     if (!vcap.isOpened())
  9.         return ;
  10.     while (true)
  11.     {
  12.         vcap >> fps_curr;
  13.         if (fps_curr.empty())
  14.             return ;
  15.         //DNN(fps_curr, my_objinfo);        //目标检测
  16.         //draw(fps_curr, my_objinfo);       //解析my_objinfo绘制框
  17.         cv::namedWindow("CPU_Video_play", cv::WINDOW_OPENGL);   //gl
  18.         cv::imshow("CPU_Video_play", fps_curr);
  19.         //pushrtsp(fps_curr,1920,1080);     //推出去rtsp
  20.         cv::waitKey(1);
  21.     }
  22.     cv::waitKey(0);
  23. }

这是一个明显的单线程线性的处理方法,很多事情在同步阻塞下是不能操作的,这也导致了中间不能插入太多要求.真正项目上是力不从心的.

下面对上面做一个改进.将解码->处理->渲染(显示)分别拆开到独立线程处理.如下图:

  1. void testMutiThread(const char* name)
  2. {
  3.     //数据结构
  4.     using MAT = cv::Mat;
  5.     using MyFrame = struct 
  6.     {
  7.         std::queue<MAT> image_data_queue;                   //解码数据
  8.         std::mutex decode_mutex;
  9.         std::queue<MAT> process_data_queue;                 //处理数据
  10.         std::mutex process_mutex;
  11.         std::queue<MAT> draw_data_queue;                    //渲染数据
  12.         std::mutex draw_mutex;
  13.     };
  14.     static MyFrame m_Frame;
  15.     std::atomic_bool brun1(true), brun2(true), brun3(true);
  16.  
  17.     auto CollectTd = [&]
  18.     {
  19.         cv::VideoCapture vcap;
  20.         MAT capture264;
  21.         vcap.open(name);           //打开文件
  22.         vcap.set(cv::CAP_PROP_POS_FRAMES, 1);
  23.         if (!vcap.isOpened())
  24.             return;
  25.         int numflag(0);
  26.         while (true)
  27.         {
  28.             std::string strnum = std::to_string(numflag++);
  29.             vcap >> capture264;     //此处模拟相机采集数据
  30.             if (!capture264.empty())
  31.             {
  32.                 std::lock_guard<std::mutex> lock_guard(m_Frame.decode_mutex);
  33.                 cout << "CollectThread image_data>> " << strnum << endl;
  34.                 m_Frame.image_data_queue.push(capture264);
  35.             }
  36.             else {
  37.                 brun1 = brun2 = brun3 = false;
  38.                 break;
  39.             }
  40.             this_thread::sleep_for(std::chrono::milliseconds(30));
  41.         }
  42.     };
  43.  
  44.     auto DecodeTd =[&]
  45.     {
  46.         while (brun1)
  47.         {
  48.             if (!m_Frame.image_data_queue.empty())
  49.             {
  50.                 MAT image_data;
  51.                 {
  52.                     std::lock_guard<std::mutex> lock_guard(m_Frame.decode_mutex);
  53.                     image_data = m_Frame.image_data_queue.front();
  54.                     m_Frame.image_data_queue.pop();
  55.                     cout << "DecodeThread image_data<< " << endl;
  56.                 }
  57.                 //从收到的数据进行解码,解码后送到处理线程
  58.                 MAT process_data = image_data;
  59.                 {
  60.                     std::lock_guard<std::mutex> lock_guard(m_Frame.process_mutex);
  61.                     cout << "DecodeThread process_data>> " << endl;
  62.                     m_Frame.process_data_queue.push(process_data);
  63.                 }
  64.             }
  65.             else
  66.                 this_thread::sleep_for(std::chrono::milliseconds(50));
  67.         }
  68.     };
  69.  
  70.     auto ProcessTd = [&]
  71.     {
  72.         while (brun2)
  73.         {
  74.             if (!m_Frame.process_data_queue.empty())
  75.             {
  76.                 MAT process_data;
  77.                 {
  78.                     std::lock_guard<std::mutex> lock_guard(m_Frame.process_mutex);
  79.                     process_data = m_Frame.process_data_queue.front();
  80.                     m_Frame.process_data_queue.pop();
  81.                     cout << "ProcessThread process_data<< " << endl;
  82.                 }
  83.                 if (!process_data.empty())
  84.                 {
  85.                     //处理图像 处理完送到渲染线程
  86.                     //ImageProcess(process_data);
  87.                     MAT draw_data = process_data;
  88.                     std::lock_guard<std::mutex> lock_guard(m_Frame.draw_mutex);
  89.                     cout << "ProcessThread draw_data>> " << endl;
  90.                     m_Frame.draw_data_queue.push(draw_data);
  91.                 }
  92.             }
  93.             else
  94.                 this_thread::sleep_for(std::chrono::milliseconds(50));
  95.         }
  96.     };
  97.  
  98.     auto DrawTd = [&]
  99.     {
  100.         cv::namedWindow("DrawThread", cv::WINDOW_OPENGL);   //gl
  101.         cv::resizeWindow("DrawThread", 800, 600);
  102.         while (brun3)
  103.         {
  104.             if (!m_Frame.draw_data_queue.empty())
  105.             {
  106.                 MAT draw_data;
  107.                 {
  108.                     std::lock_guard<std::mutex> lock_guard(m_Frame.draw_mutex);
  109.                     draw_data = m_Frame.draw_data_queue.front();
  110.                     m_Frame.draw_data_queue.pop();
  111.                     //用opencv渲染
  112.                     cv::imshow("DrawThread", draw_data);
  113.                     cv::waitKey(1);
  114.                     cout << "DrawThread draw_data<< " << endl;
  115.                 }
  116.             }
  117.             else
  118.                 this_thread::sleep_for(std::chrono::milliseconds(50));
  119.         }
  120.     };
  121.  
  122.     std::thread td1(CollectTd);
  123.     std::thread td2(DecodeTd);
  124.     std::thread td3(ProcessTd);
  125.     std::thread td4(DrawTd);
  126.     td1.join();
  127.     td2.join();
  128.     td3.join();
  129.     td4.join();
  130.     this_thread::sleep_for(std::chrono::hours(1));
  131. }

当然实际中可能还有更多的处理线程,这里我做个笔记. 所以 还是以类的形式更加直观,这里扩展一下:

采用6个线程来模拟,数据采用5个队列内部各自处理

部分参考示例代码如下:

  1. //**********************************************************************
  2. //                        C11 视频多线程框架
  3. //                                          @2021-01-04阿甘整理
  4. //**********************************************************************
  5. //通用数据结构
  6. using MAT = cv::Mat;
  7.  
  8. //共享数据结构
  9. struct MyFrame
  10. {
  11.     std::queue<MAT> image_data_queue;                   //解码数据
  12.     std::mutex decode_mutex;
  13.     std::queue<MAT> process_data_queue;                 //处理数据
  14.     std::mutex process_mutex;
  15.     std::queue<MAT> push_data_queue;                    //推流数据
  16.     std::mutex push_mutex;
  17.     std::queue<MAT> draw_data_queue;                    //渲染数据
  18.     std::mutex draw_mutex;
  19.     std::queue<MAT> tracker_data_queue;                 //跟踪数据
  20.     std::mutex tracker_mutex;
  21.     //清空方法
  22.     void clear(queue<MAT>& my_queue)
  23.     {
  24.         queue<MAT> empty_queue;
  25.         std::swap(my_queue, empty_queue);
  26.     }
  27.     void clearall(void)
  28.     {
  29.         clear(image_data_queue);
  30.         clear(process_data_queue);
  31.         clear(push_data_queue);
  32.         clear(draw_data_queue);
  33.         clear(tracker_data_queue);
  34.     }
  35. };
  36.  
  37. //线程基类
  38. class BaseThread
  39. {
  40. public:
  41.     BaseThread(MyFrame& mFrame);
  42.     virtual ~BaseThread(void);
  43.     MyFrame& m_Frame;
  44. public:
  45.     void begintk(void);
  46.     void endtk(void);
  47.     bool getstatus(void);
  48.     //线程函数用来多态重载
  49.     virtual void run(void) = 0;
  50. private:
  51.     std::atomic_bool m_status;
  52.     std::thread th;
  53. };
  54.  
  55. //采集线程
  56. class CollectTd :public BaseThread
  57. {
  58. public:
  59.     using BaseThread::BaseThread;
  60.     virtual void run(void);
  61.     void setplayMp4(string url);
  62. private:
  63.     string m_mp4;
  64.     cv::VideoCapture vcap;
  65.     MAT capture264;
  66. };
  67.  
  68. //解码线程
  69. class DecodeTd :public BaseThread
  70. {
  71. public:
  72.     using BaseThread::BaseThread;
  73.     virtual void run(void);
  74. };
  75.  
  76. //渲染线程
  77. class DrawTd :public BaseThread
  78. {
  79. public:
  80.     using BaseThread::BaseThread;
  81.     virtual void run(void);
  82. };
  83.  
  84. //处理线程
  85. class ProcessTd :public BaseThread
  86. {
  87. public:
  88.     using BaseThread::BaseThread;
  89.     virtual void run(void);
  90. };
  91.  
  92. //推流线程
  93. class PushTd :public BaseThread
  94. {
  95. public:
  96.     using BaseThread::BaseThread;
  97.     virtual void run(void);
  98. };
  99.  
  100. //跟踪线程
  101. class TrackerTd :public BaseThread
  102. {
  103. public:
  104.     using BaseThread::BaseThread;
  105.     virtual void run(void);
  106. };
  107.  
  108. //管理器
  109. class C11VideoFrame
  110. {
  111. public:
  112.     static void testrun(void)
  113.     {
  114.         C11VideoFrame mgr;
  115.         mgr.init();
  116.         mgr.beginwork();
  117.         this_thread::sleep_for(std::chrono::hours(10));
  118.     }
  119. public:
  120.     C11VideoFrame(void);
  121.     ~C11VideoFrame(void);
  122. public:
  123.     void init(void);
  124.     void beginwork(void);
  125.     void endwork(void);
  126. private:
  127.     std::unique_ptr<CollectTd> m_collect;
  128.     std::unique_ptr<DecodeTd> m_decode;
  129.     std::unique_ptr<ProcessTd> m_Process;
  130.     std::unique_ptr<PushTd> m_Push;
  131.     std::unique_ptr<DrawTd> m_Draw;
  132.     std::unique_ptr<TrackerTd> m_Tracker;
  133.     std::thread th;
  134.     MyFrame m_Frame;                    //中间数据结构
  135. };

实现部分示例:

  1. //**********************************************************************
  2. //                        C11 视频多线程框架
  3. //                                          @2021-01-04阿甘整理
  4. //**********************************************************************
  5.  
  6.  
  7. BaseThread::BaseThread(MyFrame& mFrame) : m_status(false)
  8. , m_Frame(mFrame)
  9. {
  10. }
  11.  
  12. BaseThread::~BaseThread(void)
  13. {
  14.     endtk();
  15. }
  16.  
  17. void BaseThread::begintk(void)
  18. {
  19.     if (m_status)
  20.         endtk();
  21.     m_status = true;
  22.     th = thread([&]()
  23.     {
  24.         this->run();
  25.     });
  26. }
  27.  
  28. void BaseThread::endtk(void)
  29. {
  30.     m_status = false;
  31.     if (th.joinable())
  32.         th.join();
  33. }
  34.  
  35. bool BaseThread::getstatus(void)
  36. {
  37.     return m_status;
  38. }
  39.  
  40.  
  41. void CollectTd::run(void)
  42. {
  43.     vcap.open(m_mp4);           //打开文件
  44.     vcap.set(cv::CAP_PROP_POS_FRAMES, 1);
  45.     if (!vcap.isOpened())
  46.         return;
  47.     int numflag(0);
  48.     while (getstatus())
  49.     {
  50.         std::string strnum = std::to_string(numflag++);
  51.         vcap >> capture264;     //此处模拟相机采集数据
  52.         if (!capture264.empty())
  53.         {
  54.             std::lock_guard<std::mutex> lock_guard(m_Frame.decode_mutex);
  55.             cout << "CollectThread image_data>> " << strnum << endl;
  56.             m_Frame.image_data_queue.push(capture264);
  57.         }
  58.         else
  59.             endtk();
  60.         this_thread::sleep_for(std::chrono::milliseconds(30));
  61.     }
  62. }
  63.  
  64. void CollectTd::setplayMp4(string url)
  65. {
  66.     m_mp4 = url;
  67. }
  68.  
  69. void DecodeTd::run(void)
  70. {
  71.     while (getstatus())
  72.     {
  73.         if (!m_Frame.image_data_queue.empty())
  74.         {
  75.             MAT image_data;
  76.             {
  77.                 std::lock_guard<std::mutex> lock_guard(m_Frame.decode_mutex);
  78.                 image_data = m_Frame.image_data_queue.front();
  79.                 m_Frame.image_data_queue.pop();
  80.                 cout << "DecodeThread image_data<< "  << endl;
  81.             }
  82.             //从收到的数据进行硬解码,解码后送到处理线程
  83.             MAT process_data = image_data;
  84.             {
  85.                 //cuvidxx操作
  86.                 std::lock_guard<std::mutex> lock_guard(m_Frame.process_mutex);
  87.                 cout << "DecodeThread process_data>> " << endl;
  88.                 m_Frame.process_data_queue.push(process_data);
  89.             }
  90.         }
  91.         else
  92.             this_thread::sleep_for(std::chrono::milliseconds(50));
  93.     }
  94. }
  95.  
  96. void ProcessTd::run(void)
  97. {
  98.     while (getstatus())
  99.     {
  100.         if (!m_Frame.process_data_queue.empty())
  101.         {
  102.             MAT process_data;
  103.             {
  104.                 std::lock_guard<std::mutex> lock_guard(m_Frame.process_mutex);
  105.                 process_data = m_Frame.process_data_queue.front();
  106.                 m_Frame.process_data_queue.pop();
  107.                 cout << "ProcessThread process_data<< "  << endl;
  108.             }
  109.             if (!process_data.empty())
  110.             {
  111.                 //处理图像 处理完送到渲染线程
  112.                 //ImageProcess(process_data);
  113.                 MAT draw_data = process_data;
  114.                 std::lock_guard<std::mutex> lock_guard(m_Frame.draw_mutex);
  115.                 cout << "ProcessThread draw_data>> "  << endl;
  116.                 m_Frame.draw_data_queue.push(draw_data);
  117.             }
  118.         }
  119.         else
  120.             this_thread::sleep_for(std::chrono::milliseconds(50));
  121.     }
  122. }
  123.  
  124. void DrawTd::run(void)
  125. {
  126.     cv::namedWindow("DrawThread", cv::WINDOW_OPENGL);   //gl
  127.     cv::resizeWindow("DrawThread", 800,600);
  128.     while (getstatus())
  129.     {
  130.         if (!m_Frame.draw_data_queue.empty())
  131.         {
  132.             MAT draw_data;
  133.             {
  134.                 std::lock_guard<std::mutex> lock_guard(m_Frame.draw_mutex);
  135.                 draw_data = m_Frame.draw_data_queue.front();
  136.                 m_Frame.draw_data_queue.pop();
  137.                 //用opencv渲染
  138.                 cv::imshow("DrawThread", draw_data);
  139.                 cv::waitKey(1);
  140.                 cout << "DrawThread draw_data<< " << endl;
  141.             }
  142.             //跟踪数据计算
  143.             MAT tracker_data = draw_data;
  144.             {
  145.                 std::lock_guard<std::mutex> lock_guard(m_Frame.tracker_mutex);
  146.                 cout << "DrawThread tracker_data>> " << endl;
  147.                 m_Frame.tracker_data_queue.push(tracker_data);
  148.             }
  149.             //处理推流数据
  150.             MAT push_data = draw_data;
  151.             {
  152.                 std::lock_guard<std::mutex> lock_guard(m_Frame.push_mutex);
  153.                 cout << "DrawThread push_data>> " << endl;
  154.                 m_Frame.push_data_queue.push(push_data);
  155.             }
  156.         }
  157.         else
  158.             this_thread::sleep_for(std::chrono::milliseconds(50));
  159.     }
  160. }
  161.  
  162. void PushTd::run(void)
  163. {
  164.     while (getstatus())
  165.     {
  166.         if (!m_Frame.push_data_queue.empty())
  167.         {
  168.             MAT push_data;
  169.             {
  170.                 std::lock_guard<std::mutex> lock_guard(m_Frame.push_mutex);
  171.                 push_data = m_Frame.push_data_queue.front();
  172.                 m_Frame.push_data_queue.pop();
  173.                 cout << "PushThread push_data<< " << endl;
  174.             }
  175.             //调用推送方法
  176.             if (!push_data.empty())
  177.             {
  178.                 //RtmpPuller(push_data.data, push_data.cols * push_data.rows * push_data.channels());
  179.             }
  180.         }
  181.         else
  182.             this_thread::sleep_for(std::chrono::milliseconds(50));
  183.     }
  184. }
  185.  
  186. void TrackerTd::run(void)
  187. {
  188.     while (getstatus())
  189.     {
  190.         if (!m_Frame.tracker_data_queue.empty())
  191.         {
  192.             MAT tracker_data;
  193.             {
  194.                 std::lock_guard<std::mutex> lock_guard(m_Frame.tracker_mutex);
  195.                 tracker_data = m_Frame.tracker_data_queue.front();
  196.                 m_Frame.tracker_data_queue.pop();
  197.                 cout << "TrackerThread tracker_data<< " << endl;
  198.             }
  199.             //中间跟踪器算法
  200.             //tracker(tracker_data);
  201.         }
  202.         else
  203.             this_thread::sleep_for(std::chrono::milliseconds(50));
  204.     }
  205. }
  206.  
  207.  
  208.  
  209.  
  210. C11VideoFrame::C11VideoFrame(void)
  211. {
  212. }
  213.  
  214. C11VideoFrame::~C11VideoFrame(void)
  215. {
  216. }
  217.  
  218. void C11VideoFrame::init(void)
  219. {
  220.     m_collect = unique_ptr<CollectTd>(new CollectTd(m_Frame));
  221.     m_collect->setplayMp4("cam.mp4");
  222.     m_decode = unique_ptr<DecodeTd>(new DecodeTd(m_Frame));
  223.     m_Process = unique_ptr<ProcessTd>(new ProcessTd(m_Frame));
  224.     m_Push = unique_ptr<PushTd>(new PushTd(m_Frame));
  225.     m_Draw = unique_ptr<DrawTd>(new DrawTd(m_Frame));
  226.     m_Tracker = unique_ptr<TrackerTd>(new TrackerTd(m_Frame));
  227. }
  228.  
  229. void C11VideoFrame::beginwork(void)
  230. {
  231.     m_decode->begintk();
  232.     m_Process->begintk();
  233.     m_Push->begintk();
  234.     m_Draw->begintk();
  235.     m_Tracker->begintk();
  236.     m_collect->begintk();
  237.     th = thread([&]()
  238.     {
  239.         while (true)
  240.         {
  241.             //采集线程停止 其他可以关闭了
  242.             if (!m_collect->getstatus())
  243.             {
  244.                 m_decode->endtk();
  245.                 m_Process->endtk();
  246.                 m_Push->endtk();
  247.                 m_Draw->endtk();
  248.                 m_Tracker->endtk();
  249.                 break;
  250.             }
  251.             this_thread::sleep_for(std::chrono::milliseconds(800));
  252.         }
  253.     });
  254. }
  255.  
  256. void C11VideoFrame::endwork(void)
  257. {
  258.     m_collect->endtk();
  259.     m_decode->endtk();
  260.     m_Process->endtk();
  261.     m_Push->endtk();
  262.     m_Draw->endtk();
  263.     m_Tracker->endtk();
  264. }
  265.  
  266.

最后运行效果和中间打印日志:还是以影像修复七里香来弄咯

总结:目前仅仅解决了在多线程中异步处理的思路,这里并没有对线程做干涉同步,而是用队列各自处理.在一定情况下还是有很大的折腾可自己搞的.中间的数据结构都是采用的数据拷贝这个是要根据需要自己折中选择的. 为了简单就这样了,就当是抛砖引玉,高手就见笑了.

视频分析中的多线程笔记记录(1)分享完毕.

完整下载

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/我家自动化/article/detail/300297
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号