目的

目的是为了流水处理网络视频，所有处理都为线程，每个线程单元都从TThreadRunable 继承下来，以下为TThreadRunable 基础类。从读，解码，处理算法，压缩，到发送，几个部分合成一整个流程，其中算法处理比较耗时，最后要显示到web浏览器里面，所有最后处理算法以后又直接通过websocket送到浏览器显示。远程视频这里启用的是rtsp视频。

#ifndef _TTHREAD_RUN_ABLE_H_
#define _TTHREAD_RUN_ABLE_H_#include <mutex>
#include <queue>
#include <thread>
#include <atomic>
#include <condition_variable>
using namespace std;class TThreadRunable
{private://线程thread _thread;//等待信号std::mutex _signal_mutex;std::condition_variable _cond;
protected://char  _running = false;char _stop = true;//锁定运行状态std::mutex _mutex;
public:TThreadRunable(){}virtual ~TThreadRunable(){}public:char * status(){return &_stop;}void Join(){if (_thread.joinable())_thread.join();}bool  IsStop(){return _stop == 1 ? true : false;}void WaitForSignal(){std::unique_lock<std::mutex> ul(_signal_mutex);_cond.wait(ul);}void Notify(){_cond.notify_one();}virtual int Start(){if (_stop == 0)return -1;_stop = 0;_thread = std::thread(std::bind(&TThreadRunable::Run, this));return 0;}  virtual void Stop(){_stop = 1; // true;}virtual void Run() = 0;
};
#endif

定义流程中第一个读

注意其中的SetNextProcess 函数。

#pragma once
#include <stdio.h>
#include <iostream>
#include "TThreadRunable.h"
#include "config.h"
#include <functional>
#include "TQueue.h"
#include <opencv2/opencv.hpp>
typedef enum open_type
{_type_usb,_type_rtsp,_type_rtsp_live555
}open_type;class TProcessRead :public TThreadRunable
{cv::VideoCapture cap;open_type _type;int _show = 0;vconfig * _videoconfig = NULL;func_cb_send _cs = NULL;string _name;
public:void Set(open_type type, string name,vconfig * obj){_name = name;_type = type;_videoconfig = obj;}void SetNextProcess(func_cb_send cs){_cs = cs;}Group_Mat* ReadFrame(){Group_Mat * gm = new Group_Mat();if (gm == NULL){cout << "not enough memory error" << endl;return NULL;}cap >> gm->_bgrMat;if (gm->_bgrMat.empty()){delete gm;cout << "error read frame" << endl;return NULL;}cvtColor(gm->_bgrMat, gm->_grayMat, cv::COLOR_BGR2GRAY);gm->name = _name;return gm;}int USB(){cap.open(0, cv::CAP_DSHOW);cap.set(cv::CAP_PROP_FRAME_WIDTH, _videoconfig->width);cap.set(cv::CAP_PROP_FRAME_HEIGHT, _videoconfig->height);//cap.set(cv::CAP_PROP_FPS, _videoconfig->fps); cap.open(0);cv::CAP_PROP_CHANNEL//VideoCapture cap(0);//cap.set(CAP_PROP_FRAME_WIDTH, 1920);//cap.set(CAP_PROP_FRAME_HEIGHT, 1080);//cv::CAP_CROSSBAR_INPIN_TYPE //cap.set(cv::CAP_CROSSBAR_INPIN_TYPE , 6);//Mat img;//namedWindow("test", WINDOW_NORMAL);//resizeWindow("test", 960, 640);if (!cap.isOpened()){cout << "can not open the usb" << endl;return -1;}//cap.set()double w = cap.get(cv::CAP_PROP_FRAME_WIDTH);double h = cap.get(cv::CAP_PROP_FRAME_HEIGHT);double fps = cap.get(cv::CAP_PROP_FPS);if (fps == 0)fps = 10;int delay = 1000 / fps;while (1){if (IsStop())break;Group_Mat * gm = ReadFrame();if (gm == NULL)break;if (_cs != NULL)_cs(gm);/*if (_show == 1){imshow("show", gm->_bgrMat);}*/cv::waitKey(delay);}this->Stop();return 0;};int Rtsp(){if (_videoconfig == NULL)return -1;string url = _videoconfig->url;if (cap.open(url) == false){cout << "error:rtsp url :" << endl;cout << url << " can't be open" << endl;return -1;}double w = cap.get(cv::CAP_PROP_FRAME_WIDTH);double h = cap.get(cv::CAP_PROP_FRAME_HEIGHT);double fps = cap.get(cv::CAP_PROP_FPS);if (fps == 0)fps = 20;int delay = 1000 / fps;while (1){if (IsStop())break;Group_Mat * gm = ReadFrame();if (gm == NULL)break;if (_cs != NULL)_cs(gm);/*if (_show == 1){imshow("show", gm->_bgrMat);}*/cv::waitKey(delay);}this->Stop();return 0;}void Run(){if (_type == _type_usb) {USB();}else if (_type == _type_rtsp){Rtsp();}else if (_type == _type_rtsp_live555){}}
};

先来看看我们的example.cpp 怎么进行流程的，
TProcessRead
TProcessRtsp
TProcessPdec
TProcessPbgr
TProcessSend
TProcessSpre

每个类都是从TThreadRunable继承，每个类都为线程单元类，接下去为每个对象赋值他们的回调函数
func_cb_send read_2_pdec = std::bind(&TProcessPdec::callback_recv, &process_pdec, _1);
func_cb_send read_2_pbgr = std::bind(&TProcessPbgr::callback_recv, &process_pbgr, _1);
func_cb_send read_2_send = std::bind(&TProcessSend::callback_recv, &process_send, _1);
func_cb_send read_2_serv = std::bind(&TProcessSpre::callback_recv, &process_serv, _1);

使用SetNextProcess 函数来设置下一个流程处理。

int main()
{Init();TReadConfig config;if (config.ReadConfig() != 0)return -1;//计算返回的数据//TProcessffmp process_ffmp;TProcessRead process_read;TProcessRtsp process_rtsp;TProcessPdec process_pdec;TProcessPbgr process_pbgr;TProcessSend process_send;TProcessSpre process_serv;func_cb_send read_2_pdec = std::bind(&TProcessPdec::callback_recv, &process_pdec, _1);func_cb_send read_2_pbgr = std::bind(&TProcessPbgr::callback_recv, &process_pbgr, _1);func_cb_send read_2_send = std::bind(&TProcessSend::callback_recv, &process_send, _1);func_cb_send read_2_serv = std::bind(&TProcessSpre::callback_recv, &process_serv, _1);//config startif (config.use_usb == 1){string name;vconfig *obj = config.GetUsbFirst(name);if(obj!=NULL)process_read.Set(_type_usb,name,obj);}else if (config.use_rtsp == 1){string name;vconfig * obj = config.GetRtspFirst(name);if (obj != NULL){//process_ffmp.Set(obj->url.c_str(), obj->tcp, name.c_str());process_rtsp.Set(obj->url.c_str(), obj->tcp, name.c_str());process_pdec.Set(name.c_str());}}process_send.Set(NULL, ws_callback);//config overif (config.use_usb == 1){process_read.SetNextProcess(read_2_pbgr);}else if (config.use_rtsp == 1){process_rtsp.SetNextProcess(read_2_pdec);process_pdec.SetNextProcess(read_2_pbgr);}//else if (config.user_file == 1)//{// process_ffmp.SetNextProcess(read_2_pbgr);//}process_pbgr.SetNextProcess(read_2_send);process_send.SetNextProcess(read_2_serv);uint16_t port =(uint16_t)config.wsserver_port;if (port == 0)port = 9090;//在9090端口上等待数据cout << "ws server list at " << port << endl;process_serv.Start(port);process_send.Start();cout << "process algorithm start"<<endl;process_pbgr.Start();cout << "process read start" << endl;process_pdec.Start();cout << "process pdec start" << endl;if (config.use_usb == 1){process_read.Start();process_read.Join();}else if(config.use_rtsp == 1){process_rtsp.Start();process_rtsp.Join();}return 0;
}

算法处理单元

以下为线程处理单元的run线程处理函数

void TProcessPbgr::Run()
{while (1){if (IsStop())break;WaitForSignal();Group_Mat * data = _group.Pop();while (data != NULL){if (IsStop())break;if (_cs != NULL){param_data * d = new param_data();//传递名字d->name = data->name;//准备以jpg发送出去d->type = _enumjpg;d->rgb = data->_bgrMat;_algorithm.articulation(data->_grayMat, d->articulate);_algorithm.colorException(data->_bgrMat, d->c_cast, d->c_da, d->c_db);_algorithm.brightnessException(data->_grayMat, d->b_cast, d->b_da);//msk = _algorithm.DectectorMsk(data->_grayMat,40,200);_cs(d);}delete data;data = _group.Pop();}}
}

总结

这种方式比较好理解，改进也有很多方式，线程是宝贵的资源，该节省的地方还是要节省，每一路都使用很多线程是比较耗费的。后面会进行改进，除了普通算法，为AI处理推理也提供了一种思路

一种流水线方式处理远程视频的方案相关推荐

秀场直播的四种实现方式，让我们从方案架构进行分析
秀场互动直播是 RTC 技术应用的常见场景,虽然主播 PK 的业务逻辑不算复杂,但由于在标准直播模式和主播 PK 模式的切换过程中容易产生卡顿.黑屏等现象,为了在优雅实现业务逻辑的同时,最大程度缓解类 ...
i.MX6ULL - 远程视频监控方案实现（nginx-rtmp流媒体服务器、ffmpeg推流）
i.MX6ULL - 远程视频监控配置(nginx-rtmp流媒体服务器.ffmpeg推流) 目录 i.MX6ULL - 远程视频监控配置(nginx-rtmp流媒体服务器.ffmpeg推流) 1.前 ...
一种基于深度学习的视频编码方案
为了设计一套新的编码方法,我们需要考虑以下几个方面: 1.编码效率:编码后的数据要尽可能地短,以减少存储和传输的成本. 2.解码速度:解码过程应尽可能快,以减小解码成本. 3.适应性:编码方法应该能够 ...
网易云信助力长沙银行打造远程视频银行 | 字母点评数字化先锋案例
关于"数字化先锋"系列数字化转型是迈向数字经济社会的重要一步,是数字化技术与传统产业融合的重要进程.在数字化的探索中,不乏先锋者的身影,这些"数字化先锋"凭借 ...
FME教程：表格数据分组统计总数的三种实现方式
在FME中进行表格数据分组统计总数的三种方式,实现类似SQL中分组统计总数的功能. 一.业务场景现在有图1所示的一份Excel表格数据,要求统计每个村的户数,如图2所示. 图1 Excel表格数据 ...
php抓取图片curl,php获取远程图片的两种 CURL方式和sockets方式获取远程图片
php获取远程图片的两种:CURL方式和sockets方式获取远程图片,需要的朋友可以参考下. 方式1:sockets $a = "http://jb51.net/content/uploa ...
【数据库视频】七种连接方式
数据表的查询与管理只是针对数据库中的一个表格进行的查询管理,如果现在我们想要同时的看到两个数据表中的数据的或,需要怎么实现?答案是:使用多连接的方式进行查询标题中说了SQL中有七种连接的方式,那么具 ...
江苏：研究生毕业答辩可采用远程视频方式
近日,江苏省新型冠状病毒感染的肺炎疫情防控工作领导小组学校防控组印发<关于做好开学前及春季学期研究生教育教学管理工作的指导意见>.<意见>就开学前研究生教育管理.春季学期教学计 ...
网易云信联手长沙银行，远程视频银行系统助力数字化转型
数字经济时代,5G.人工智能.大数据.云计算等新技术正改变着以银行为代表的金融行业的业务流程.产品模式和客户服务方式,办理金融业务已不再是只依靠物理网点和电话客服,线上APP服务.零接触服务等形式日渐 ...

一种流水线方式处理远程视频的方案

目的

定义流程中第一个读

算法处理单元

总结

一种流水线方式处理远程视频的方案相关推荐

最新文章

热门文章