[TOC] ### 線程池的應用場景 ***** #### 任務隊列 ``` #pragma once #ifndef THREAD_POOL_H #define THREAD_POOL_H #include <vector> #include <queue> #include <thread> #include <atomic> #include <condition_variable> #include <future> #include <functional> #include <stdexcept> #define MAX_THREAD_NUM 256 //線程池,可以提交變參函數或拉姆達表達式的匿名函數執行,可以獲取執行返回值 //不支持類成員函數, 支持類靜態成員函數或全局函數,Opteron()函數等 // void wait (unique_lock<mutex>& lck, Predicate pred); //std::condition_variable.wait()，只有當 pred 條件為 false 時調用 wait() 才會阻塞當前線程 //std::future 可以用來獲取異步任務的結果，因此可以把它當成一種簡單的線程間同步的手段 //std::forward就可以保存參數的左值或右值特性。 //轉移構造函數和轉移賦值函數，也就是把一個左值引用當做右值引用來使用，怎么做呢？標準庫提供了函數 std::move，這個函數以非常簡單的方式將左值引用轉換為右值引用。 class threadpool { using Task = std::function<void()>; // 線程池 std::vector<std::thread> pool; // 任務隊列 std::queue<Task> tasks; // 同步 std::mutex m_lock; // 條件阻塞 std::condition_variable cv_task; // 是否關閉提交 std::atomic<bool> stoped; //空閑線程數量 std::atomic<int> idlThrNum; public: inline threadpool(unsigned short size = 4) :stoped{ false } { idlThrNum = size < 1 ? 1 : size; for (size = 0; size < idlThrNum; ++size) { //初始化線程數量 pool.emplace_back( [this] { // 工作線程函數 while (!this->stoped) { std::function<void()> task; { // 獲取一個待執行的 task std::unique_lock<std::mutex> lock{ this->m_lock };// unique_lock 相比 lock_guard 的好處是：可以隨時 unlock() 和 lock() this->cv_task.wait(lock, [this] { return this->stoped.load() || !this->tasks.empty(); } ); // wait 直到有 task if (this->stoped && this->tasks.empty()) return; task = std::move(this->tasks.front()); // 取一個 task this->tasks.pop(); } idlThrNum--; task(); idlThrNum++; } } ); } } inline ~threadpool() { stoped.store(true); cv_task.notify_all(); // 喚醒所有線程執行 for (std::thread& thread : pool) { //thread.detach(); // 讓線程“自生自滅” if (thread.joinable()) thread.join(); // 等待任務結束， 前提：線程一定會執行完 } } public: // 提交一個任務 // 調用.get()獲取返回值會等待任務執行完,獲取返回值 // 有兩種方法可以實現調用類成員， // 一種是使用 bind： .commit(std::bind(&Dog::sayHello, &dog)); // 一種是用 mem_fn： .commit(std::mem_fn(&Dog::sayHello), &dog) template<class F, class... Args> auto commit(F&& f, Args&&... args) ->std::future<decltype(f(args...))> { if (stoped.load()) // stop == true ?? throw std::runtime_error("commit on ThreadPool is stopped."); using RetType = decltype(f(args...)); // typename std::result_of<F(Args...)>::type, 函數 f 的返回值類型 auto task = std::make_shared<std::packaged_task<RetType()> >( std::bind(std::forward<F>(f), std::forward<Args>(args)...) ); // wtf ! std::future<RetType> future = task->get_future(); { // 添加任務到隊列 std::lock_guard<std::mutex> lock{ m_lock };//對當前塊的語句加鎖 lock_guard 是 mutex 的 stack 封裝類，構造的時候 lock()，析構的時候 unlock() tasks.emplace( [task]() { // push(Task{...}) (*task)(); } ); } cv_task.notify_one(); // 喚醒一個線程執行 return future; } //空閑線程數量 int idlCount() { return idlThrNum; } }; #endif ``` ### 基本使用方法 ***** ``` #include <iostream> #include "ThreadPool.h" int main() { // create thread pool with 4 worker threads ThreadPool pool(4); // enqueue and store future auto result = pool.enqueue([](int answer) { return answer; }, 42); // get result from future, print 42 std::cout << result.get() << std::endl; } ``` 另一個例子 ``` #include <iostream> #include "ThreadPool.h" void func() { std::this_thread::sleep_for(std::chrono::milliseconds(100)); std::cout<<"worker thread ID:"<<std::this_thread::get_id()<<std::endl; } int main() { ThreadPool pool(4); while(1) { pool.enqueue(fun); } } ```