多线程模板应用实现（实践学习笔记）

出处：B站码出名企路

个人笔记：因为是跟着b站的教学视频以及文档初步学习，可能存在诸多的理解有误，对大家仅供借鉴，参考，然后是B站up阳哥的视频，我是跟着他学。大家有兴趣的可以到b站搜索。加油，一起学习。我的问题，大家如果看见，希望可以提出指正，谢谢大家。

应用场景

多线程的应用场景非常多，常见的有：

1. 网络通信：在网络通信应用中，一般需要同时处理多个请求，如果使用单线程模式，会阻塞其他请求，造成性 能瓶颈，因此使用多线程可以提高并发处理能力。

2. 数据库操作：在数据库操作中，有时需要同时对多个数据表进行操作，使用多线程可以提高处理效率。

3. 图像处理：在图像处理应用中，需要对多个图像进行处理，在单线程模式下，处理速度会很慢，使用多线程可 以提高处理速度。

4. 游戏开发：在游戏开发中，常常需要同时处理多个任务，比如处理游戏画面、物理效果、声音效果等，使用多 线程可以提高游戏的运行速度和流畅度。

5. 并行计算：在科学计算领域中，常常需要对大量数据进行处理和计算，使用多线程可以将计算任务划分到多个 线程中进行，从而提高计算速度。

总之，多线程在提高程序性能、响应性和资源利用率方面有着广泛的应用。然而，需要注意在多线程编程中处理线程同步、共享数据等问题，以确保程序的正确性和稳定性。

图解结构

模块拆解

第一步：StateSubmitor耗时内容处理类

此处并没有很多具体实现，因为要结合业务。比如耗时处理逻辑

  class StateSubmitor    {public:explicit StateSubmitor(const std::string& str);~StateSubmitor();//submit: 提交到队列中//const std::string& content 内容，包括海量数据void submit(const std::string& content);//content可任意//flush: 将队列中的所有状态信息发往远程收集端//具体的业务逻辑void flush();private:StateSubmitor(const StateSubmitor&) = delete;StateSubmitor& operator=(const StateSubmitor&) = delete;};

    void StateSubmitor::submit(const std::string& content){/*@ 对 content的耗时处理逻辑*/}

第二步：NodeMonitor线程启动类

//节点监控, 监控任务的发生, 业务的产生. 多线程同步等控制逻辑的封装class NodeMonitor{public:~NodeMonitor();static NodeMonitor* instance();void start();void shutdown();bool init();private:NodeMonitor();NodeMonitor(const NodeMonitor&) = delete;NodeMonitor& operator=(const NodeMonitor&) = delete;void stateInfo(const std::string& strs);void ThreadFunc();                         //消费者线程入口函数bool shutdown_;                            //开关   std::mutex mutex_;                         std::thread thread_;                       //消费者线程std::condition_variable cond_;//queuestd::queue<std::string> task_queue_;       //任务队列std::unique_ptr<StateSubmitor> submitor_;  //unique_ptr管理submitor对象};}

具体实现，这里才是多线程同步互斥的重点部分，核心，利用任务队列做缓冲容器，解耦合。使得生产者线程和消费者线程之间的耦合度降低，生产者只管将任务放入任务队列，然后即可返回，无需等待消费者处理。消费者只管从任务队列中拿取任务处理。大大提高效率。通过缓存大大减低了生产者和消费者之间的耦合程度。

生活场景：快递驿站，快递小哥就是生产者，我们就是消费者。快递驿站就是容器队列。

 //析构一般独立一个函数NodeMonitor::~NodeMonitor(){this->shutdown();//做资源释放等等操作}//创建线程安全的单例//call_once 确保多线程下仅仅创建一个NodeMonitor对象NodeMonitor* NodeMonitor::instance(){static NodeMonitor* instance = nullptr;static std::once_flag flag;  std::call_once(flag, [&]{instance = new (std::nothrow) NodeMonitor();});return instance; }//线程启动void NodeMonitor::start(){//创建消费者thread_ = std::thread(&NodeMonitor::ThreadFunc, this);//启动生产者if (!init()){return;}}//生产者函数bool NodeMonitor::init(){submitor_.reset(new StateSubmitor("lyy")); //创建submitor/*@ 不断地填充stateInfo@ 如果是实际应用场景可能会采取轮询, 或者是event事件触发, 此处阳哥按照最简单的塞入文本信息作为事件(任务)*/while (true){stateInfo("lxk");}return true;}//填入需要的信息 <=> push任务void NodeMonitor::stateInfo(const std::string& strs){std::unique_lock<std::mutex> lock(mutex_);task_queue_.push(strs); //生产, 塞入任务cond_.notify_one();     //通知消费}//线程销毁void NodeMonitor::shutdown(){std::unique_lock<std::mutex> lock(mutex_);shutdown_ = true;cond_.notify_all();if (thread_.joinable()){thread_.join();}}//消费者函数void NodeMonitor::ThreadFunc(){while (!shutdown_){std::unique_lock<std::mutex> lock(mutex_);cond_.wait(lock, [this]{return shutdown_ || !task_queue_.empty();});if (shutdown_){break;}std::string str = task_queue_.front();task_queue_.pop();lock.unlock();submitor_->submit(str);//提交状态信息}}

具体案例

消息队列作业实现

#include <iostream>
#include <queue>
#include <mutex>
#include <thread>
#include <memory>
#include <condition_variable>
#include <string>
#include <chrono>namespace XX
{class MessageQueue {//封装消息队列类public:void push(const std::string& message); std::string pop(); bool empty();private:std::mutex mutex_; //互斥锁, 保障互斥操作std::condition_variable cond_; //通知, 保障同步std::queue<std::string> msg_queue_;  //容器};class StateSubmitor {//消息处理类, 业务处理, 管理消息队列public:explicit StateSubmitor(MessageQueue& msg_queue);~StateSubmitor();void submit(const std::string& content); //提交状态信息并将其添加到队列中void flush();  //flush: 将队列中的所有状态信息发往远程收集端, 清空处理所有消息.private:StateSubmitor(const StateSubmitor &) = delete;StateSubmitor &operator=(const StateSubmitor &) = delete;private:MessageQueue& msg_queue_;  //消息队列};// 节点监控, 监控任务的发生, 业务的产生. 多线程同步等控制逻辑的封装class NodeMonitor {public:~NodeMonitor();static NodeMonitor *instance();void start();void shutdown();bool init();private:NodeMonitor();void ProducerThreadFunc(); //线程函数void ConsumerThreadFunc(); //线程函数NodeMonitor(const NodeMonitor &) = delete;NodeMonitor &operator=(const NodeMonitor &) = delete;private:std::thread producer_thread_; //生产者线程，不停的往消息队列塞入监控到的用户状态信息消息.static int count_;std::unique_ptr<StateSubmitor> submitor_;MessageQueue msg_queue_; //消息队列std::thread consumer_thread_;//消费者线程, 不停的从消息队列中抽出消息进行处理bool shutdown_;              //开关};
}namespace XX {int NodeMonitor::count_ = 0;//初始化void MessageQueue::push(const std::string& message) {std::unique_lock<std::mutex> lock(mutex_);msg_queue_.push(message);//塞入消息cond_.notify_one();//通知消费}std::string MessageQueue::pop() {std::unique_lock<std::mutex> lock(mutex_);cond_.wait(lock, [this]{//等待消息到来return !empty();});std::string msg = msg_queue_.front();//拿到消息msg_queue_.pop();return msg;}bool MessageQueue::empty() {return msg_queue_.empty();}StateSubmitor::StateSubmitor(MessageQueue& msg_queue): msg_queue_(msg_queue) {}  void StateSubmitor::submit(const std::string& content) {//提交状态信息消息的业务操作std::cout << "消息为: " << content << std::endl;//将业务状态消息push到消息队列中msg_queue_.push(content);}void StateSubmitor::flush() {//清空所有消息}StateSubmitor::~StateSubmitor() {this->flush();}NodeMonitor::NodeMonitor():shutdown_(false){}NodeMonitor::~NodeMonitor(){this->shutdown();//释放资源...操作}void NodeMonitor::ProducerThreadFunc() {while (!shutdown_) { //不断生产std::this_thread::sleep_for(std::chrono::milliseconds(3000));std::string msg = "消息";msg += std::to_string(count_);count_ ++;submitor_->submit(msg);}}NodeMonitor* NodeMonitor::instance(){static NodeMonitor* instance = nullptr;static std::once_flag flag;  std::call_once(flag, [&]{instance = new (std::nothrow) NodeMonitor();});return instance; }void NodeMonitor::ConsumerThreadFunc() {while (!shutdown_) { //不断消费std::this_thread::sleep_for(std::chrono::milliseconds(2000));std::string msg = msg_queue_.pop();//弹出一条消息std::cout << "处理了: " << msg << std::endl;}}void NodeMonitor::start() {init();}void NodeMonitor::shutdown() {shutdown_ = true;}bool NodeMonitor::init() {submitor_.reset(new StateSubmitor(msg_queue_)); //创建submitor//创建生产者，消费者线程并且joinproducer_thread_ = std::thread(&NodeMonitor::ProducerThreadFunc, this);consumer_thread_ = std::thread(&NodeMonitor::ConsumerThreadFunc, this);producer_thread_.join();consumer_thread_.join();return true;}
}int main() {XX::NodeMonitor::instance()->start();return 0;
}