Linux线程池的实现

线程池

线程池是线程的一种使用模式线程池linux，是管理和利用多线程处理多任务的一种方法。

解决的问题

当线程过多时，会带来调度开销，进而影响整体性能；

线程池维护多个线程，等待管理者分配可并发执行的任务；避免处理短时间任务时，创建和销毁线程的代价。

线程池可以保证内核的充分利用，防止过分调度。线程池中的线程数量取决于当前可用的并发处理器，内存，网络，套接字等。

应用场景

固定线程数的线程池的模拟实现

线程池就是大量的线程+任务队列

当任务队列为空时，线程阻塞

当有任务进入任务队列时，唤醒线程处理

//线程池 = 大量线程 + 任务队列
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
#include
using namespace std;
#define MAX_THREAD 5      //最大线程数
#define MAX_QUEUE 10      //安全队列的最大个数
typedef void(*handler_t)(int);
//任务类
class MyTask{
  private:
      int _data;
      handler_t _handler;
  public:
      MyTask(int data, handler_t handler)
      :_data(data)
      ,_handler(handler)
      {}
      void Run()
      {
        return _handler(_data);			//任务
      }
};
class ThreadPool{
  private:
      vector<thread> _thr_list;   //存储线程的操作句柄
      queue<MyTask> _qTask;       //任务队列
      
      int _capacity;              
      pthread_mutex_t _mutex;
      pthread_cond_t _cond_con;   //消费者的条件变量
      pthread_cond_t _cond_pro;   //生产者的条件变量
      int _thr_max;       //最大线程数
      int _thr_cur;       //当前线程数
      bool _quit_flag;    //退出线程的标志
      
      void thr_start()          //线程的入口
      {
        while(1)
        {
          pthread_mutex_lock(&_mutex);
          while(_qTask.empty())
          {
            if(_quit_flag == true)			//退出标志
            {
              cout<<"thread exit"<<pthread_self()<<endl;
              pthread_mutex_unlock(&_mutex);
              _thr_cur--;
              return ;
            }
            pthread_cond_wait(&_cond_con,&_mutex);		//无任务时阻塞
          }
		  //处理任务
          MyTask tt = _qTask.front();
          _qTask.pop();
          pthread_mutex_unlock(&_mutex);	
          pthread_cond_signal(&_cond_pro);
          tt.Run();
        }
        return ;
      }
  public:
    ThreadPool(int maxq = MAX_QUEUE, int maxt = MAX_THREAD)
      :_thr_max(maxt)
       ,_capacity(maxq)
       ,_thr_list(maxt)
       ,_quit_flag(false)
     {
       pthread_mutex_init(&_mutex,NULL);
       pthread_cond_init(&_cond_con,NULL);
      pthread_cond_init(&_cond_pro,NULL);
     }
    ~ThreadPool()
   {
       pthread_mutex_destroy(&_mutex);
       pthread_cond_destroy(&_cond_con);
       pthread_cond_destroy(&_cond_pro);
   }
   bool PoolInit()
   {
     _thr_cur = 0;
     for(int i=0; i<_thr_max; i++)
     {
       _thr_list[i] = thread(&ThreadPool::thr_start, this);
       _thr_cur++;
       _thr_list[i].detach();		//线程处理完任务后直接回收
     }
     return true;
   }
   bool AddTask(MyTask& tt)
   {
     pthread_mutex_lock(&_mutex);
     while(_qTask.size() == _capacity)		//当任务队列满时
     {
       pthread_cond_wait(&_cond_pro,&_mutex);
     }
     _qTask.push(tt);
     pthread_mutex_unlock(&_mutex);
     pthread_cond_signal(&_cond_con);		//唤醒等待线程
     return true;
   }
   //线程池销毁
   bool PoolStop()
   {
     pthread_mutex_lock(&_mutex);
     _quit_flag = true;
     pthread_mutex_unlock(&_mutex);
     
     while(_thr_cur > 0)
     {
      pthread_cond_broadcast(&_cond_con);	//广播唤醒所有线程，将阻塞线程回收
      usleep(1000);
     }
     return true;
   }
};

测试用例：

void test(int data)
{
  srand(time(NULL));
  int nesc = rand()%5;
  cout<<"thread:"<<pthread_self()<<"~~~Thread data:"<<data<<" sleep:"<<nesc<<" seconds"<<endl;
  sleep(nesc);
  return ;
}
int main()
{
  ThreadPool pool;
  pool.PoolInit();
  for(int i=0; i<10; i++)
  {
    MyTask tt(i,test);
	pool.AddTask(tt);
  }
  pool.PoolStop();
  return 0;
}

（编辑：成都站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!