用C++简单写的线程池

测试用例在后面： 参考地址 Thread.h

/* Datetime:2020/8/21 By LiuZhenguang */ #ifndef _THREADPOOL__H #define _THREADPOOL__H #include <vector> #include <iostream> #include <mutex> #include <thread> #include <deque> #include <condition_variable> #include <functional> class ThreadPool{ public: ThreadPool(int init_size=3); ~ThreadPool(); int threadSize; typedef std::function<void()> Task;//std::function的实例可以对任何可以调用的目标实体进行存储、复制、和调用操作 void stop(); void addTask(const Task& t); private: ThreadPool(const ThreadPool& t);//声明为private禁止拷贝构造函数被调用 const ThreadPool operator=(const ThreadPool& t); bool isStart(){ return this->m_isStart; } void start(); void ThreadLoop(); Task take();//从任务队列取任务 typedef std::vector<std::thread*> Threads;//线程队列结构描述 typedef std::deque<Task> Tasks;//一个std::function<void()>的队列 Threads m_threads;//实际线程队列 Tasks m_tasks; std::mutex m_mutex; std::condition_variable m_cond; bool m_isStart; }; #endif

Thread.cpp

#include "ThreadPool.h" #include <iostream> #include <assert.h> #include <sstream> //构造函数 ThreadPool::ThreadPool(int init_size) : threadSize(init_size), m_mutex(), m_cond(), m_isStart(false){ start(); } //将线程初始化并启动 void ThreadPool:: start(){ assert(m_threads.empty());//如果括号内不成立，则程序立即终止 m_isStart = true; m_threads.reserve(threadSize);//设置线程队列容量为3; for (int i = 0; i < threadSize; ++i){ m_threads.push_back(new std::thread(std::bind(&ThreadPool::ThreadLoop, this)));//将线程与线程循环绑定 } std::cout << "线程池初始化成功" << std::endl; } //线程循环 void ThreadPool:: ThreadLoop(){ while (m_isStart) { Task task = take();//如果为空将阻塞11111111 if (task){ std::cout << "取1任务成功" << std::endl; task();//任务执行 } } } //取任务 ThreadPool::Task ThreadPool::take(){//这个是自定义类型，需要作用域解析符 std::unique_lock<std::mutex> lock(m_mutex);//锁住任务队列，防止和addTask冲突 while (m_tasks.empty() && m_isStart){//若队列为空，`wait`将解锁互斥量，阻塞到本行,等待addTask的notify all()111111 m_cond.wait(lock);//这个lock是上面一行m_mutex所在的类对象 } Task task; //Task::size_type size = m_tasks.size();//报错 int size = m_tasks.size(); if (!m_tasks.empty() && m_isStart) { task = m_tasks.front(); m_tasks.pop_front();//队列取任务，FIFO assert(size - 1 == m_tasks.size());//没取走则报错 /*if (TaskQueueSize_ > 0) { cond2.notify_one(); }*/ } return task; } //添加任务 void ThreadPool:: addTask(const Task& t){ std::unique_lock<std::mutex> lock(m_mutex);//加锁 m_tasks.push_back(t); std::cout << "任务添加成功" << std::endl; m_cond.notify_all();//唤醒取任务 } //关闭线程池 void ThreadPool:: stop(){ std::unique_lock<std::mutex> m_lock(m_mutex);//stop()函数结束后可以自动解锁，可以使用lock_guard就可以了 m_isStart = false; m_cond.notify_all();//其他线程使用wait，然后阻塞了，直到notify_all()唤醒所有其他线程 for (auto i : m_threads){//存的就是指针 i->join();//让主线程的等待每一个子线程完成，然后主线程再继续执行，确保关闭前，所有任务都能完成 delete i; } m_threads.clear();//线程池清空 std::cout << "线程池关闭"<<std::endl; } //析构函数 ThreadPool::~ThreadPool(){ if (m_isStart) m_isStart = false; stop(); }

测试用例：

#include <iostream> #include <chrono> //延时 #include <condition_variable> #include "ThreadPool.h" std::mutex g_mutex; void testFunc() { for (int i = 0; i < 100; ++i){ std::cout << std::this_thread::get_id() << std::endl; std::this_thread::sleep_for(std::chrono::seconds(1)); } } int main() { ThreadPool thread_pool; for (int i = 0; i < 5; i++) thread_pool.addTask(testFunc); system("pause"); return 0; }``