c++11线程池

  1 #pragma once
  2 
  3 #include <future>
  4 
  5 #include <vector>
  6 
  7 #include <atomic>
  8 
  9 #include <queue>
 10 
 11 #include <thread>
 12 
 13 #include <mutex>
 14 
 15 namespace std {
 16 
 17 //线程池最大容量,应尽量设小一点
 18 #define THREADPOOL_MAX_NUM 16
 19 
 20 class ThreadPool
 21 {
 22 public:
 23     ThreadPool(unsigned short size = 1) { AddThread(size); }
 24     ~ThreadPool()
 25     {
 26         if (_run.load())
 27         {
 28             Close();
 29         }
 30     }
 31 
 32     void Close()
 33     {
 34         _run.store(false);
 35         //唤醒所有线程执行
 36         _task_cv.notify_all();
 37         for (thread &th : _pool)
 38         {
 39             if (th.joinable())
 40                 th.join();
 41         }
 42     }
 43 
 44     //提交一个任务，
 45     template<class F, class... Args>
 46     auto commit(F&& f, Args&&... args) ->future<decltype(f(args...))>
 47     {
 48         if (!_run)
 49             throw runtime_error("commit on ThreadPool is stop.");
 50         // typename std::result_of<F(Args...)>::type, 函数 f 的返回值类型
 51         using RetType = decltype(f(args...));
 52         //把函数入口及参数打包
 53         auto task = make_shared<packaged_task<RetType()>>(bind(forward<F>(f), forward<Args>(args)...));
 54 
 55         future<RetType> future = task->get_future();
 56         {
 57             lock_guard<mutex> lock{ _lock };
 58             _tasks.emplace([task]() {(*task)(); });
 59         }
 60 #ifdef THREADPOOL_AUTO_GROW if (_id1ThrNum < 1 && _pool.size() < THREADPOOL_MAX_NUM) AddThread(1); #endif _task_cv.notify_one(); return future; }
 61 
 62 int IdlCount() { return _id1ThrNum; }
 63     int BusyCount() { return _pool.size(); }
 64 
 65     void AddThread(unsigned short size)
 66     {
 67         for (; _pool.size() < THREADPOOL_MAX_NUM && size > 0; --size)
 68         {
 69             _pool.emplace_back([this] {
 70                 while (_run.load())
 71                 {
 72                     Task task;
 73                     {
 74                         unique_lock<mutex> lock{ _lock };
 75                         _task_cv.wait(lock, [this]
 76                         {
 77                             return !_run.load() || !_tasks.empty();
 78                         });
 79                         if (!_run.load() && _tasks.empty())
 80                             return;
 81                         task = move(_tasks.front());
 82                         _tasks.pop();
 83                     }
 84                     _id1ThrNum--;
 85                     task();
 86                     _id1ThrNum++;
 87                 }
 88             });
 89             _id1ThrNum--;
 90         }
 91     }
 92 
 93 public:
 94     //定义类型
 95     using Task = std::function<void()>;
 96     //线程池
 97     vector<thread> _pool;
 98     //锁
 99     mutex _lock;
100     //任务队列
101     queue<Task> _tasks;
102     //条件阻塞
103     condition_variable _task_cv;
104     //线程是否在执行
105     atomic<bool> _run{ true };
106     //空闲线程
107     atomic<int> _id1ThrNum{ 0 };
108 };
109 }