当前位置: 首页 > news >正文

C++简易线程池

原理说明:

1. 线程池创建时,指定线程池的大小thread_size。当有新的函数任务通过函数addFunction ()添加进来后,其中一个线程执行函数。一个线程一次执行一个函数。如果函数数量大与线程池数量,则后来的函数等待。

2. 线程池内部有个容器m_functions 来存储待执行的函数。函数执行后从队列中移除。

3.  stopAll()函数会停止线程池。

ThreadPool.h

//ThreadPool.h
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <future>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>class ThreadPool {
public:static ThreadPool* getInstance(size_t thread_size = 1);		//默认线程池大小void addFunction(std::function<void()> task);				//添加需要执行的函数~ThreadPool();void stopAll(bool immediately);						//停止线程池, immediately:true立即停止, immediately:false等待当前线程函数执行完后停止。
private:ThreadPool(size_t thread_size);private:void workerThreadHandler();std::vector<std::thread> m_workers;					//线程容器std::queue<std::function<void()>> m_functions;			//待执行的函数容器std::mutex m_queue_mutex;std::condition_variable m_condition;bool m_stop;										//线程池停止状态std::thread m_wakeTimerThread;std::mutex m_timer_mutex;
};

ThreadPool.cpp

#include "ThreadPool.h"
#include <chrono>
#include <memory>
#include <thread>
#include <pthread.h>
#include <iostream>
using namespace std;ThreadPool* ThreadPool::getInstance(size_t thread_size)
{static std::mutex m_lock;static std::shared_ptr<ThreadPool> m_instance=nullptr;if (nullptr == m_instance){m_instance.reset(new ThreadPool(thread_size));}return m_instance.get();
}
ThreadPool::ThreadPool(size_t thread_size) : m_stop(false){m_workers.reserve(thread_size);for (size_t i=0; i<thread_size; ++i){m_workers.emplace_back([this, i](){	//创建线程池中的线程workerThreadHandler();});}//辅助线程,每隔一段时间发送一次唤醒,防止线程阻塞m_wakeTimerThread = std::thread([this](){	for(;!this->m_stop;){std::unique_lock<std::mutex> lock(this->m_timer_mutex);std::this_thread::sleep_for(std::chrono::milliseconds(2000));pthread_testcancel();m_condition.notify_all();std::cout<<"wake up"<<std::endl;}});std::cout<<__func__<<std::endl;
}//线程循环函数,循环查询函数容器是否为空,不为空则读取一个函数并执行。
void ThreadPool::workerThreadHandler()	
{for (;!this->m_stop;){std::function<void()> task;{std::unique_lock<std::mutex> lock(this->m_queue_mutex);std::cout<<"tasks begin size:"<<this->m_functions.size()<<" stop:"<<m_stop<<std::endl;if (!m_stop && m_functions.empty()){this->m_condition.wait(lock);}pthread_testcancel();	//作为线程的终止点if (this->m_stop){return;}if (this->m_functions.empty()){continue;}task = std::move(this->m_functions.front());this->m_functions.pop();std::cout<<std::this_thread::get_id() <<" tasks end size:"<<this->m_functions.size()<<std::endl;}task();std::cout<<__func__<<" end task"<<std::endl;std::this_thread::sleep_for(std::chrono::milliseconds(200));}
}void ThreadPool::addFunction(std::function<void()> task)	//添加待执行的函数
{std::unique_lock<std::mutex> lock(m_queue_mutex);if (m_stop){return;}m_functions.emplace(std::move(task));m_condition.notify_one();
}ThreadPool::~ThreadPool()
{{std::unique_lock<std::mutex> lock(m_queue_mutex);m_stop = true;}m_condition.notify_all();for (std::thread &worker: m_workers){worker.join();}m_wakeTimerThread.join();std::cout<<__func__<<std::endl;
}void ThreadPool::stopAll(bool immediately)	//停止线程, immediately:true立即停止
{this->m_stop = true;if (immediately){for (std::thread &worker: m_workers){pthread_cancel(worker.native_handle());}pthread_cancel(m_wakeTimerThread.native_handle());}
}

测试程序main.cpp

#include <iostream>
#include <chrono>
#include <mutex>
#include "ThreadPool.h"using namespace std;static std::mutex m_mutex;
void ProcessFunc111()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc222()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc333()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc444()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc555()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc666()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}int main()
{ThreadPool::getInstance(5);ThreadPool::getInstance()->addFunction([](){ProcessFunc111();});ThreadPool::getInstance()->addFunction([](){ProcessFunc222();});ThreadPool::getInstance()->addFunction([](){ProcessFunc333();});ThreadPool::getInstance()->addFunction([](){ProcessFunc444();});ThreadPool::getInstance()->addFunction([](){ProcessFunc555();});ThreadPool::getInstance()->addFunction([](){ProcessFunc666();});getchar();return 0;
}

执行结果:

tasks begin size:0 stop:0
ThreadPooltasks begin size:
0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
140563906656000 tasks end size:2
ProcessFunc111 begin
140563898263296 tasks end size:4
ProcessFunc222 begin
140563881477888 tasks end size:3
ProcessFunc333 begin
140563743504128 tasks end size:2
ProcessFunc444 begin
140563889870592 tasks end size:1
ProcessFunc555 begin
wake up
ProcessFunc111 end
workerThreadHandler end task
tasks begin size:1 stop:0
140563906656000 tasks end size:0
ProcessFunc666 begin
ProcessFunc222 end
wake up
workerThreadHandler end task
ProcessFunc333ProcessFunc555ProcessFunc444 endendworkerThreadHandler end taskworkerThreadHandler end taskend
workerThreadHandler end task
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
wake up
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
ProcessFunc666 end
workerThreadHandler end task
tasks begin size:0 stop:0
wake up
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
...

测试程序2,调用线程池停止程序

#include <iostream>
#include <chrono>
#include <mutex>
#include "ThreadPool.h"using namespace std;static std::mutex m_mutex;
void ProcessFunc111()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc222()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc333()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc444()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc555()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc666()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}int main()
{ThreadPool::getInstance(5);ThreadPool::getInstance()->addFunction([](){ProcessFunc111();});ThreadPool::getInstance()->addFunction([](){ProcessFunc222();});ThreadPool::getInstance()->addFunction([](){ProcessFunc333();});ThreadPool::getInstance()->addFunction([](){ProcessFunc444();});ThreadPool::getInstance()->addFunction([](){ProcessFunc555();});ThreadPool::getInstance()->addFunction([](){ProcessFunc666();});std::this_thread::sleep_for(std::chrono::seconds(1));std::cout<<"stop all "<<std::endl;ThreadPool::getInstance()->stopAll(false);getchar();return 0;
}

执行结果:

tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:ThreadPool0 stop:0tasks begin size:0 stop:0
140190941017856 tasks end size:5
ProcessFunc111 begin
140190932625152 tasks end size:4
ProcessFunc222 begin
140190966195968 tasks end size:3
ProcessFunc333 begin
140190949410560 tasks end size:2
ProcessFunc444 begin
140190957803264 tasks end size:1
ProcessFunc555 begin
stop all
wake up
ProcessFunc111 end
workerThreadHandler end task
ProcessFunc444 end
workerThreadHandler end task
ProcessFunc333 end
workerThreadHandler end task
ProcessFunc222 end
workerThreadHandler end task
ProcessFunc555 end
workerThreadHandler end task

测试程序3,立即停止线程池

#include <iostream>
#include <chrono>
#include <mutex>
#include "ThreadPool.h"using namespace std;static std::mutex m_mutex;
void ProcessFunc111()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc222()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc333()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc444()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc555()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(4));std::cout<<__func__<<" end"<<std::endl;
}void ProcessFunc666()
{std::cout<<__func__<<" begin"<<std::endl;std::this_thread::sleep_for(std::chrono::seconds(3));std::cout<<__func__<<" end"<<std::endl;
}int main()
{ThreadPool::getInstance(5);ThreadPool::getInstance()->addFunction([](){ProcessFunc111();});ThreadPool::getInstance()->addFunction([](){ProcessFunc222();});ThreadPool::getInstance()->addFunction([](){ProcessFunc333();});ThreadPool::getInstance()->addFunction([](){ProcessFunc444();});ThreadPool::getInstance()->addFunction([](){ProcessFunc555();});ThreadPool::getInstance()->addFunction([](){ProcessFunc666();});std::this_thread::sleep_for(std::chrono::seconds(1));std::cout<<"stop all "<<std::endl;ThreadPool::getInstance()->stopAll(true);getchar();return 0;
}

执行结果:

tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
tasks begin size:0 stop:0
ThreadPool
139831215929088 tasks end size:5
ProcessFunc111 begin
tasks begin size:5 stop:0
139831199143680 tasks end size:4
ProcessFunc222 begin
139831224321792 tasks end size:3
ProcessFunc333 begin
139831207536384 tasks end size:2
ProcessFunc444 begin
139831232714496 tasks end size:1
ProcessFunc555 begin
stop all

http://www.lryc.cn/news/270111.html

相关文章:

  • 【MATLAB】PSO粒子群优化LSTM(PSO_LSTM)的时间序列预测
  • 产品经理学习-怎么写PRD文档
  • 第3课 获取并播放音频流
  • Spark编程实验四:Spark Streaming编程
  • Flink去重计数统计用户数
  • 力扣:62. 不同路径(动态规划,附python二维数组的定义)
  • 2022年全球运维大会(GOPS深圳站)-核心PPT资料下载
  • 8868体育助力意甲罗马俱乐部 迪巴拉有望付出
  • java设计模式实战【策略模式+观察者模式+命令模式+组合模式,混合模式在支付系统中的应用】
  • 小程序wx:if 和hidden的区别?
  • 自动驾驶学习笔记(二十三)——车辆控制模型
  • Linux Shell 015-文本双向覆盖重定向工具tee
  • 【PyQt】(自定义类)QIcon派生,更易用的纯色Icon
  • 【mysql】数据处理格式化、转换、判断
  • 深入探索Java中的UDP网络通信机制
  • List常见方法和遍历操作
  • 【基础篇】一、认识JVM
  • DrGraph原理示教 - OpenCV 4 功能 - 颜色空间
  • 听GPT 讲Rust源代码--src/tools(36)
  • 学生数据可视化与分析工具 vue3+flask实现
  • uni-app condition启动模式配置
  • 网大为卸任腾讯CXO;Midjourney 1 月训练视频模型;2023年马斯克赚了7700亿
  • 据报道,微软的下一代 Surface 笔记本电脑将是其首款真正的“人工智能 PC”
  • Springer build pdf乱码
  • k8s之kudeadm
  • NModbus-一个C#的Modbus协议库实现
  • Altium Designer20中遇到的问题和解决办法记录
  • flask web学习之flask与http(二)
  • 基于Python的电商手机数据可视化分析和推荐系统
  • 汽车制造厂批量使用成华制造弹簧平衡器