学习基于C++0x的threadpool简单实现

  1 #include <iostream>
  2 #include <map>
  3 #include <vector>
  4 #include <mutex>
  5 #include <future>
  6 #include <thread>
  7 #include <memory>
  8 #include <condition_variable>
  9 #include <functional>
 10 #include <queue>
 11 #include <stdexcept>
 12 
 13 class ThreadPool
 14 {
 15 public:
 16     ThreadPool(int size);
 17 
 18     template<class F, class... Args>
 19     auto enqueue(F&& f, Args... args) -> std::future<typename std::result_of<F(Args...)>::type>;
 20 
 21     ~ThreadPool();
 22 
 23 private:
 24     std::vector<std::thread> workers;
 25     std::queue<std::function<void()>> tasks;
 26 
 27     std::mutex _mutex;
 28     std::condition_variable _cond_var;
 29     bool stop;
 30 };
 31 
 32 ThreadPool::ThreadPool(int size):stop(false)
 33 {
 34     for(int i = 0; i < size; i++)
 35     {
 36         workers.emplace_back(
 37             [this] ()
 38             {
 39                 // 每个线程循环工作
 40                 for (;;)
 41                 {
 42                     std::function<void()> task;
 43                     {
 44                         std::unique_lock<std::mutex> lock(this->_mutex);
 45                         // 当队列为空 或者 没有停止pool时，会阻塞在这里，直到有任务
 46                         this->_cond_var.wait(lock, [this] ()
 47                         {
 48                             return this->stop || !this->tasks.empty();
 49                         });
 50                         // 对应析构中 notify_all 此时 stop 为 true
 51                         if (this->stop || this->tasks.empty())
 52                         {
 53                             return;
 54                         }
 55                         task = std::move(this->tasks.front());
 56                         this->tasks.pop();
 57                     }
 58                     task();
 59                 }
 60             }
 61         );
 62     }
 63 }
 64 
 65 /**
 66  * 这个看起来很复杂的定义是
 67  * 定义了函数 enqueue， 接收两个模板类型的参数， F类型 和 Args类型
 68  * F类型是函数，Args类型是参数列表
 69  * 返回值类型是 std::future<>， future返回的具体类型是 typename std::result_of<F(Args...)>::type ， auto 是占位符
 70  * typename std::result_of<F(Args...)>::type 的意思是，方法 F(Args...) 的返回值的类型
 71  * 所以就是 std::future<F(Args...)> enqueue(F, Args...)
 72  * 可能出现的问题，就是塞入了太多的任务，导致队列内存爆掉
 73  */
 74 template<class F, class... Args>
 75 auto ThreadPool::enqueue(F&& f, Args... args) -> std::future<typename std::result_of<F(Args...)>::type>
 76 {
 77     // get return type
 78     using return_type = typename std::result_of<F(Args...)>::type;
 79     // create task
 80     // std::make_shared 产生了一个指针，指向了一个方法，方法是一个 std::packaged_task<> 包装的task，
 81     // 类型的签名是 return_type() 就是无参数，返回值类型 return_type
 82     auto task = std::make_shared<std::packaged_task<return_type()>>(
 83         std::bind(std::forward<F>(f), std::forward<Args>(args)...)
 84     );
 85     // get futre
 86     std::future<return_type> ret = task->get_future();
 87     {
 88         std::unique_lock<std::mutex> lock(_mutex);
 89         if (stop)
 90         {
 91             std::cout << "thread pool has stopped";
 92             throw std::runtime_error("pool stopped");
 93         }
 94         tasks.emplace([task] ()
 95         {
 96             (*task)();
 97         });
 98     }
 99     // notify_one 之后不一定能立刻执行，有空闲的线程才执行，没有就在队列里等待
100     _cond_var.notify_one();
101     return ret;
102 }
103 
104 ThreadPool::~ThreadPool()
105 {
106     // 停止pool前要锁住
107     {
108         std::unique_lock<std::mutex> lock(_mutex);
109         stop = true;
110     }
111     // 要通知多有的线程都停止工作，并join掉所有的线程
112     _cond_var.notify_all();
113     for (auto& item: workers)
114     {
115         item.join();
116     }
117 }
118 
119 int main()
120 {
121     ThreadPool _pool(4);
122     /////////////////////////////////////////////////////////////
123     auto result1 = _pool.enqueue([](int ans)
124     {
125         return ans;
126     }, 100);
127     std::cout << "ret:" << result1.get() << std::endl;
128 
129     /////////////////////////////////////////////////////////////
130     auto result2 = _pool.enqueue([](char ans)
131     {
132         return ans;
133     }, ‘a‘);
134     std::cout << "ret:" << result2.get() << std::endl;
135 
136     /////////////////////////////////////////////////////////////
137     auto result3 = _pool.enqueue([](float ans)
138     {
139         return ans;
140     }, 1.23);
141     std::cout << "ret:" << result3.get() << std::endl;
142 
143     /////////////////////////////////////////////////////////////
144     auto result4 = _pool.enqueue([](std::string ans)
145     {
146         return ans;
147     }, "hello world");
148     std::cout << "ret:" << result4.get() << std::endl;
149 
150     return 0;
151 }