/* * 线程池的核心思想是，程序启动时会创建多个线程并运行，每个线程的 * 线程函数所做的工作就是循环等待 "任务队列" 有没有新的数据，如果有，则从队列里取出来，执行里面的回调函数. * 额外有其他的线程不停的往任务队列里塞数据，如果线程均处于忙碌状态，多余的任务就在队列里等待*/#include 
    
     #include 
     
      #include 
      
       #include 
       
        #include 
        
         #include 
         
          using namespace std;int g_task_id = 1;struct TaskInfo;typedef int (*TaskFunc)(TaskInfo* task);// 锁std::mutex con_var_mu_;std::condition_variable cond_var_;std::mutex task_queue_mu_;struct TaskInfo{ int task_id; TaskFunc callback; int param1; int param2;};void* thread_func(void* param);class ThreadPool{ public: ThreadPool(int num); ~ThreadPool();private: ThreadPool(); ThreadPool(const ThreadPool& rhs); ThreadPool& operator=(const ThreadPool& rhs);public: void Init(); void AddTask(TaskInfo* task);public: static ThreadPool& Instance() // 线程池直接使用该单例即可 { static ThreadPool pool = ThreadPool(2); return pool; }public: // 简单起见，统一都设为public，不封装了 bool stoped_; std::vector
          
            threadids_; std::queue
           
             tasks_;};// *****************************************impliment********************************* //void ThreadPool::Init(){ cout << "pool init..." << endl;}ThreadPool::ThreadPool(int num) : stoped_(false){ for (int i = 0; i < num; i++) { pthread_t *cur_t = new pthread_t; pthread_create(cur_t, NULL, thread_func, this); threadids_.push_back(cur_t); cout << "init a thread:" << *cur_t << endl; } cout << "*****************************************************************" << endl;}ThreadPool::~ThreadPool(){ // 通知线程退出 stoped_ = true; //std::unique_lock
            
              lk(con_var_mu_); // 因为lk不析构，导致con_var_mu_锁得不到释放，线程函数里的获取锁就会死锁，这里必须注掉 con_var_mu_.lock(); cond_var_.notify_all(); con_var_mu_.unlock(); // 等待线程退出 for (int i = 0; i < threadids_.size(); i++) { pthread_join(*threadids_[i], NULL); } for (int i = 0; i < threadids_.size(); i++) { delete threadids_[i]; } // 销毁任务队列的数据 task_queue_mu_.lock(); while (!tasks_.empty()) { TaskInfo* task = tasks_.front(); delete task; tasks_.pop(); } task_queue_mu_.unlock();}void ThreadPool::AddTask(TaskInfo* task){ task_queue_mu_.lock(); tasks_.push(task); task_queue_mu_.unlock(); std::unique_lock
             
               lk(con_var_mu_); // cond_var_.notify_all(); //会导致多个线程同时被通知到？ 惊群效应 cond_var_.notify_one();}// 待执行的任务对应的回调函数int task_func(TaskInfo* task){ cout << "task id:" << task->task_id << " execute..." << endl; cout << "param1: " << task->param1 << " plus param2: " << task->param2 << ", result: " << task->param1 + task->param2 << endl;}// 线程的入口函数，负责执行任务，执行完成从队列取任务，如果没有任务，则通过wait()等待void* thread_func(void* param){ while (true) { std::unique_lock
              
                lk(con_var_mu_); while (ThreadPool::Instance().tasks_.size() == 0 && !ThreadPool::Instance().stoped_) { cond_var_.wait(lk); } if (ThreadPool::Instance().stoped_) { cout << "thread:" << pthread_self() << " stoped." << endl; pthread_exit(NULL); } // have task cout << "thread:" << pthread_self() << " get a task..." << endl; task_queue_mu_.lock(); TaskInfo* task = ThreadPool::Instance().tasks_.front(); ThreadPool::Instance().tasks_.pop(); task_queue_mu_.unlock(); task->callback(task); delete task; } return NULL;}int main(){ ThreadPool::Instance().Init(); for (int i = 0; i < 10; i++) { TaskInfo* task = new TaskInfo(); task->callback = task_func; task->param1 = i; task->param2 = i * 5; task->task_id = g_task_id++; ThreadPool::Instance().AddTask(task); } sleep(2);}复制代码