安徽网站,长尾关键词排名系统,seo网站关键词优化哪家好,购物网站设计目的多线程数据读写顺序处理 一个典型的生产者-消费者模型#xff0c;在这个模型中#xff0c;多个工作线程并行处理从共享队列中获取的数据#xff0c;并将处理结果以保持原始顺序的方式放入另一个队列。
多线程处理模型#xff0c;具体细节如下#xff1a;
1.数据:数据里必…多线程数据读写顺序处理 一个典型的生产者-消费者模型在这个模型中多个工作线程并行处理从共享队列中获取的数据并将处理结果以保持原始顺序的方式放入另一个队列。
多线程处理模型具体细节如下
1.数据:数据里必须有个递增的标识符和一个结束标识(ending) 2. 读队列(安全队列)用于存放待处理的数据。 处理线程每个线程都是一个死循环读数据-处理数据-写数据,它们被编号为1、2、3、4等。这些线程负责从读队列中取出数据进行处理。线程的结束判断数据里的ending为true. 结果聚合处理完成后判断数据的递增的标识符是否为全局的递增的标识符如果相等 继续执行。以保持数据的一致性。 写队列(安全队列)用于处理好的数据按照读的顺序写入,写入数据到输出队列的顺序是保持一致的。
自定义设计多线程模版
#include queuestable.h
#ifndef QUEUESTABLE_H
#define QUEUESTABLE_H#include queue
#include mutex
#include condition_variable
#include functional
#include atomic//定义一个线程安全的队列template typename T
class QueueStable
{
public:QueueStable() default;QueueStable(const QueueStableT) delete;QueueStable operator(const QueueStableT) delete;QueueStable(unsigned int max_size){m_max_size max_size;}//设置存放数据的最大容量void set_max_size(unsigned int max_size){m_max_size max_size;}void push(T value){std::unique_lockstd::mutex lock(m_mutex);//队列大于20就阻塞m_cv.wait(lock, [] { return m_queue.size() m_max_size; }); //为 true 继续执行否则解锁-》等待。唤醒后加锁//入队m_queue.push(std::move(value));//m_queue.push(value);//解锁lock.unlock();m_cv.notify_one(); //唤醒另一个}T pop(){//加锁std::unique_lockstd::mutex lock(m_mutex);m_cv.wait(lock, [] { return !m_queue.empty(); }); //为 true 继续执行否则解锁-》等待。唤醒后加锁//出队T data std::move(m_queue.front());//T data m_queue.front();m_queue.pop();//解锁lock.unlock();m_cv.notify_one(); //唤醒另一个return data;}T front() const{std::lock_guardstd::mutex lock(m_mutex);return m_queue.front();}// bool empty() const// {// std::lock_guardstd::mutex lock(m_mutex);// return m_queue.empty();// }// int size()// {// //加锁// std::lock_guardstd::mutex lock(m_mutex);// return m_queue.size();// }private:std::queueT m_queue;std::mutex m_mutex;std::condition_variable m_cv;unsigned int m_max_size 1;
};//包数据
struct BaseData
{//递增的标识符 sequence_numberint64_t sequence_number ; //记录当前位置保证数据顺序一致bool ending false; //true 代表结尾};//多线程处理
template typename T,typename P
class MultiThreadProcessing
{static_assert(std::is_base_ofBaseData, T::value, T must be derived from BaseData!);public:enum class ThreadMode{Detach, // 分离Join // 阻塞};MultiThreadProcessing(){m_ending false;m_thread_mode ThreadMode::Join;}//设置上下文结构使用的数据 - 必须设置void set_contexts(const std::vectorstd::shared_ptrP contexts){//生成用于线程的结构for(int i0; icontexts.size(); i){auto context m_contexts.emplace_back();context.id i;context.ext_context contexts[i];}}//设置运行的处理函数 - 必须设置void set_execute_function(const std::functionvoid(const std::shared_ptrT,const std::shared_ptrP) function){m_execute_function function;}//设置线程模式void set_thread_mode(MultiThreadProcessing::ThreadMode thread_mode ThreadMode::Join){m_thread_mode thread_mode;}//多线程同步检测void start(const std::shared_ptrQueueStablestd::shared_ptrT read_queue,const std::shared_ptrQueueStablestd::shared_ptrT write_queue){//检查处理函数是否可调用if (!m_execute_function){// 打印异常信息std::cerr MultiThreadProcessing: Invalid to execute function. std::endl;throw std::runtime_error(Invalid to execute function.);return;}std::shared_ptrDataPacket data_packer read_queue-pop();//取出第一个包查看序号 不处理m_sequence_number data_packer-sequence_number;//先找到第一个包的序号m_ckeck_sequence_number data_packer-sequence_number;write_queue-push(data_packer);m_sequence_number;m_ckeck_sequence_number;//循环开启线程for(auto context : m_contexts){//开启线程context.thread std::make_sharedstd::thread([this,context,read_queue,write_queue](){std::cout context.id: context.id Started. std::endl;for(;;){// 读数据 逻辑//加锁m_read_mutex.lock();//为true末尾标志不在继续直接结束if(m_ending){m_read_mutex.unlock();break;}context.temp_data read_queue-pop();//检测是否为递增1if(m_ckeck_sequence_number.load() context.temp_data-sequence_number){m_ckeck_sequence_number;}else{m_ending true;// 打印异常信息std::cerr MultiThreadProcessing: The sequence number must be incremented by one. std::endl;throw std::runtime_error(The sequence number must be incremented by one.);break;//异常}//true 包的末尾結束if(context.temp_data-ending){m_ending true;std::unique_lockstd::mutex lock(m_write_mutex);m_write_cond_var.wait(lock, [] { return m_sequence_number.load() context.temp_data-sequence_number; }); //为 true 继续执行否则解锁-》等待。唤醒后加锁//写入包write_queue-push(context.temp_data);//解锁lock.unlock();m_write_cond_var.notify_all(); //唤醒所有break;}m_read_mutex.unlock();//自定义函数处理数据m_execute_function(context.temp_data,context.ext_context);//写数据逻辑std::unique_lockstd::mutex lock(m_write_mutex);m_write_cond_var.wait(lock, [] { return m_sequence_number.load() context.temp_data-sequence_number; }); //为 true 继续执行否则解锁-》等待。唤醒后加锁//写入包write_queue-push(context.temp_data);m_sequence_number;//解锁lock.unlock();m_write_cond_var.notify_all(); //唤醒所有}std::cout context.id: context.id Finished. std::endl;});}//阻塞线程for(auto context : m_contexts){if (context.thread-joinable()){if(m_thread_mode ThreadMode::Join){context.thread-join();}else if (m_thread_mode ThreadMode::Detach){context.thread-detach();}}}}private:struct S{std::shared_ptrT temp_data;std::shared_ptrstd::thread thread;uint32_t id;std::shared_ptrP ext_context; //外部的自定义数据};std::vectorS m_contexts;//每个线程的临时数据用于多线程临时//结束标志std::atomicbool m_ending false;std::mutex m_read_mutex;//读锁std::mutex m_write_mutex;//写锁std::condition_variable m_write_cond_var;//写条件变量 同步数据顺序std::atomicint64_t m_sequence_number;//保证数据顺序std::atomicint64_t m_ckeck_sequence_number;//检查数据序号若数据序号不是递增为1抛出异常std::functionvoid(const std::shared_ptrT,const std::shared_ptrP) m_execute_function; //每个数据单次处理函数MultiThreadProcessing::ThreadMode m_thread_mode ThreadMode::Join;
};#endif // QUEUESTABLE_H
QueueStable 安全队列类 BaseData 基本数据 MultiThreadProcessing多线程模版类 ,处理继承基本数据BaseData的结构体
使用 QueueStable类确保数据的读写正常MultiThreadProcessing多线程处理数据保证数据先读的先写。
举例下面实现openVINO yolov8推理代码使用上面的多线程模版
yolov8_2.h
#ifndef YOLOV8_2_H
#define YOLOV8_2_H#include filterbase.h
#include queuestable.h
//包数据
struct DataPacket : BaseData
{AVMediaType av_media_type AVMEDIA_TYPE_UNKNOWN;//记录当前的索引,分辨是音频还是视频字幕等std::shared_ptrAVPacketPtr packet; //存放包//记录av_media_type类型数据如果数据没解码数据在packet中std::vectorstd::shared_ptrAVFramePtr frame_vector;
};class YOLOV8_2 : public FilterBase
{
public:struct Config{float nms_threshold;float score_threshold;std::string model_path;std::string bin_path {};std::string properties GPU.0;uint32_t image_interval 2; // 处理图像的间隔每 image_interval处理一次};struct Detection{int class_id;std::string class_name; //类型名float confidence;//置信度cv::Rect box; // 矩形框位置};struct InferContext{ov::InferRequest request;uint32_t image_interval; // 处理图像的间隔每 image_interval处理一次};YOLOV8_2(){}YOLOV8_2(const Config config);~YOLOV8_2(){}//实现基类的纯虚函数 start 是个接口用于实现多态基类不做实现void start(const std::shared_ptrQueueStablestd::shared_ptrDataPacket read_queue,const std::shared_ptrQueueStablestd::shared_ptrDataPacket write_queue) override;//多线程同步检测void detect(const std::shared_ptrQueueStablestd::shared_ptrDataPacket read_queue,const std::shared_ptrQueueStablestd::shared_ptrDataPacket write_queue);protected:void initial();//预处理ov::Tensor preprocess(const cv::Mat frame, cv::Mat pre_frame);//后处理void postprocess(cv::Mat frame, const ov::Tensor output_tensor);cv::Mat letterbox(const cv::Mat input_image, const cv::Size target_size, const cv::Scalar fill_color cv::Scalar(0, 0, 0), float* m_ratio nullptr,int* m_top_offset nullptr,int* m_left_offset nullptr);private:Config m_config; // 参数float m_ratio; // 原图与模型输入图 缩放比例int m_top_offset;int m_left_offset;ov::CompiledModel m_compiled_model;MultiThreadProcessingDataPacket,InferContext m_multi_thread;std::vectorstd::shared_ptrInferContext m_infer_request_vector; //推理列表用于多路推理uint32_t m_infer_request_size;
};
#endif // YOLOV8_2_H#include yolov8_2.h
const std::vectorstd::string coconame { person,bicycle,car,motorcycle,airplane,bus,train,truck,boat,traffic light,fire hydrant,stop sign,parking meter,bench,bird,cat,dog,horse,sheep,cow,elephant,bear,zebra,giraffe,backpack,umbrella,handbag,tie,suitcase,frisbee,skis,snowboard,sports ball,kite,baseball bat,baseball glove,skateboard,surfboard,tennis racket,bottle,wine glass,cup,fork,knife,spoon,bowl,banana,apple,sandwich,orange,broccoli,carrot,hot dog,pizza,donut,cake,chair,couch,potted plant,bed,dining table,toilet,tv,laptop,mouse,remote,keyboard,cell phone,microwave,oven,toaster,sink,refrigerator,book,clock,vase,scissors,teddy bear,hair drier,toothbrush };YOLOV8_2::YOLOV8_2(const YOLOV8_2::Config config)
{this-m_infer_request_size 12;m_config config;initial();
}//多线程同步检测
void YOLOV8_2::detect(const std::shared_ptrQueueStablestd::shared_ptrDataPacket read_queue,const std::shared_ptrQueueStablestd::shared_ptrDataPacket write_queue)
{std::functionvoid(const std::shared_ptrDataPacket,const std::shared_ptrInferContext) lambda [this](const std::shared_ptrDataPacket data_packer,const std::shared_ptrInferContext context){if(data_packer-av_media_type AVMEDIA_TYPE_VIDEO !data_packer-frame_vector.empty()){for(const auto frame : data_packer-frame_vector){if(frame-get_number() % context-image_interval ! 0){continue;}// 创建 cv::Mat 对象注意这里直接使用 AVFrame 的数据cv::Mat image cv::Mat(frame-get()-height, frame-get()-width, CV_8UC3, frame-get()-data[0], frame-get()-linesize[0]);if (!image.empty()){cv::Mat pre_image;//临时存储//预处理ov::Tensor input_tensor preprocess(image,pre_image);//开始推理context-request.set_input_tensor(input_tensor);context-request.infer();//等待完成处理结果const ov::Tensor output_tensor context-request.get_output_tensor();this-postprocess(image, output_tensor);}}}};m_multi_thread.set_contexts(m_infer_request_vector);m_multi_thread.set_execute_function(lambda);m_multi_thread.start(read_queue,write_queue);
}
void YOLOV8_2::start(const std::shared_ptrQueueStablestd::shared_ptrDataPacket read_queue,const std::shared_ptrQueueStablestd::shared_ptrDataPacket write_queue)
{detect(read_queue,write_queue);
}
void YOLOV8_2::initial()
{//创建推理引擎 ieov::Core core;//读取模型/** std::shared_ptrov::Model model core.read_model(this-onnx_path);* brief从IR / ONNX / PDPD / TF / TFLite文件格式读取模型。* param model_path模型的路径。* param bin_path数据文件的路径。* 对于IR格式(*.bin) :* 如果bin_path为空将尝试读取与XML同名的bin文件* *如果没有找到同名的bin文件将加载无权重的IR。* 对于以下文件格式不使用bin_path参数:* ONNX格式(*.onnx)* *PDPD(*.pdmodel)* *TF(*.pb)* *TFLite(*.tflite)* 返回一个模型。*/std::shared_ptrov::Model model core.read_model(m_config.model_path,m_config.bin_path);ov::preprocess::PrePostProcessor ppp ov::preprocess::PrePostProcessor(model);ppp.input().tensor().set_element_type(ov::element::u8).set_layout(NHWC).set_color_format(ov::preprocess::ColorFormat::BGR)/*.set_spatial_static_shape(640, 640) //640*640 yolov8输入大小*/;//ppp.input().tensor().set_shape(ov::PartialShape({ 1,640,640,3 }));//自定义输入大小,确保和模型大小同样ppp.input().preprocess().convert_layout(NCHW).convert_element_type(ov::element::f32).convert_color(ov::preprocess::ColorFormat::RGB).scale({ 255, 255, 255 });// .scale({ 112, 112, 112 });//ppp.input().preprocess().resize(ov::preprocess::ResizeAlgorithm::RESIZE_NEAREST, 640, 640);//ppp.input().model().set_layout(NCHW);ppp.output().postprocess().convert_element_type(ov::element::f32);//ppp.output().tensor().set_element_type(ov::element::f32);model ppp.build();this-m_compiled_model core.compile_model(model,m_config.properties);//创建推理请求for(size_t i0;im_infer_request_size;i){std::shared_ptrInferContext sh std::make_sharedInferContext();sh-request m_compiled_model.create_infer_request();sh-image_interval m_config.image_interval;m_infer_request_vector.push_back(sh);}
}// Letterbox 缩放函数
/** input_image 输入原图像* target_size 目标图像大小* fill_color 填充颜色* m_ratio 缩放比例* m_top_offset 缩放的图像 在目标图像中的 y 位置* m_left_offset 缩放的图像 在目标图像中的 x 位置*/
cv::Mat YOLOV8_2::letterbox(const cv::Mat input_image, const cv::Size target_size, const cv::Scalar fill_color, float* m_ratio,int* m_top_offset,int* m_left_offset)
{//输出图像cv::Mat output_image(target_size, input_image.type(), fill_color);//输入图像和输出图像 高度和宽度 都相等直接复制返回if(input_image.cols output_image.cols input_image.rows output_image.rows){input_image.copyTo(output_image);//获取比例if (m_ratio){*m_ratio 1.0;}if (m_top_offset){*m_top_offset 0;}if (m_left_offset){*m_left_offset 0;}return output_image;}float r 0.0;cv::Rect dest_rect;//输入图像宽 图像高宽对齐高至中if (input_image.cols input_image.rows){// 宽缩放 m_ratio 那么高也要缩放 m_ratior static_castfloat(input_image.cols) / output_image.cols;int new_rows static_castint(input_image.rows / r);dest_rect cv::Rect(0, (output_image.rows - new_rows) / 2, output_image.cols, new_rows);//dest_rect cv::Rect(0, 0, output_image.cols, new_rows);}else{// 高缩放 m_ratio 那么宽也要缩放 m_ratior static_castfloat(input_image.rows) / output_image.rows;int new_cols static_castint(input_image.cols / r);dest_rect cv::Rect((output_image.cols - new_cols) / 2, 0, new_cols, output_image.rows);}//获取比例if (m_ratio){*m_ratio r;}if (m_top_offset){*m_top_offset dest_rect.y;}if (m_left_offset){*m_left_offset dest_rect.x;}cv::resize(input_image, output_image(dest_rect), dest_rect.size(), cv::INTER_LINEAR);return output_image;
}
//预处理
ov::Tensor YOLOV8_2::preprocess(const cv::Mat frame, cv::Mat pre_frame)
{//预处理const ov::Shape shape m_compiled_model.input().get_shape();//shape 对应 ppp.input().tensor().set_element_type(ov::element::u8).set_layout(NHWC) 中 NHWCpre_frame letterbox(frame, cv::Size(shape.at(2), shape.at(1)), cv::Scalar(100, 100, 100), m_ratio, m_top_offset, m_left_offset);uchar* input_data pre_frame.data;return ov::Tensor(m_compiled_model.input().get_element_type(), m_compiled_model.input().get_shape(), input_data);
}
//后处理
void YOLOV8_2::postprocess(cv::Mat frame, const ov::Tensor output_tensor)
{std::vectorcv::Rect boxes;std::vectorint class_ids;std::vectorfloat confidences;const ov::Shape output_shape output_tensor.get_shape();const int out_rows output_shape.at(1);const int out_cols output_shape.at(2);const cv::Mat det_output(out_rows, out_cols, CV_32F, (float*)output_tensor.datafloat());CHECK(det_output.cols 8400)CHECK(det_output.rows 84)//找到所有符合的 类别 矩形置信度for (int i 0; i det_output.cols; i){const cv::Mat classes_scores det_output.col(i).rowRange(4, 84);cv::Point class_id_point;double score;cv::minMaxLoc(classes_scores, nullptr, score, nullptr, class_id_point);//阈值大于0.25 认为检测出结果//if (score 0.3){//坐标const float x det_output.atfloat(0, i);const float y det_output.atfloat(1, i);const float w det_output.atfloat(2, i);const float h det_output.atfloat(3, i);cv::Rect box;box.x static_castint(x);box.y static_castint(y);box.width static_castint(w);box.height static_castint(h);boxes.push_back(box);class_ids.push_back(class_id_point.y);confidences.push_back(score);}}std::vectorint nms_result;//nms 去重找到最优数据cv::dnn::NMSBoxes(boxes, confidences, m_config.score_threshold, m_config.nms_threshold, nms_result);std::vectorDetection output;for (int i 0; i nms_result.size(); i){Detection result;int idx nms_result.at(i);result.class_id class_ids.at(idx);result.confidence confidences.at(idx);result.class_name coconame.at(result.class_id) std::to_string(result.confidence).substr(0, 4);result.box.width boxes.at(idx).width * this-m_ratio;result.box.height boxes.at(idx).height * this-m_ratio;result.box.x (boxes.at(idx).x - 0.5 * boxes.at(idx).width - this-m_left_offset) * this-m_ratio ;result.box.y (boxes.at(idx).y - 0.5 * boxes.at(idx).height - this-m_top_offset) * this-m_ratio ;output.push_back(result);}//绘制for (int i 0; i output.size(); i){auto detection output.at(i);auto box detection.box;auto class_string detection.class_name;float xmax box.x box.width;float ymax box.y box.height;//生成随机颜色// 获取当前系统时间作为种子auto current_time std::chrono::system_clock::now().time_since_epoch().count();// 使用随机种子创建 RNG 对象cv::RNG rng(current_time);cv::Scalar color cv::Scalar(rng.uniform(100, 256),rng.uniform(100, 256),rng.uniform(100, 256));// Detection boxcv::rectangle(frame, cv::Point(box.x, box.y), cv::Point(xmax, ymax), color, 2);// Detection box textcv::Size textSize cv::getTextSize(class_string, cv::FONT_HERSHEY_DUPLEX, 1, 2, 0);cv::Rect textBox(box.x, box.y - 40, textSize.width 10, textSize.height 20);cv::rectangle(frame, textBox, color, cv::FILLED);cv::putText(frame, class_string, cv::Point(box.x 5, box.y - 10), cv::FONT_HERSHEY_DUPLEX, 1, cv::Scalar(0, 0, 0));}
}
这个例子是ffmpeg读取的数据用openVINO推理实现对每张图片进行实时推理推理后的数据按照顺序写入显示保证数据顺序一致性。
