基於C++11的執行緒池(threadpool),簡潔且可以帶任意多的引數（轉）

阿新 • • 發佈：2018-12-14

咳咳。C++11 加入了執行緒庫，從此告別了標準庫不支援併發的歷史。然而 c++ 對於多執行緒的支援還是比較低階，稍微高階一點的用法都需要自己去實現，譬如執行緒池、訊號量等。執行緒池(thread pool)這個東西，在面試上多次被問到，一般的回答都是：“管理一個任務佇列，一個執行緒佇列，然後每次取一個任務分配給一個執行緒去做，迴圈往復。” 貌似沒有問題吧。但是寫起程式來的時候就出問題了。

廢話不多說，先上實現，然後再囉嗦。(dont talk, show me ur code !)

程式碼實現

  1 #pragma once
  2 #ifndef THREAD_POOL_H
  3 #define 
 THREAD_POOL_H
  4 
  5 #include <vector>
  6 #include <queue>
  7 #include <thread>
  8 #include <atomic>
  9 #include <condition_variable>
 10 #include <future>
 11 #include <functional>
 12 #include <stdexcept>
 13 
 14 namespace std
 15 {
 16 #define 
  MAX_THREAD_NUM 256
 17 
 18 //執行緒池,可以提交變參函式或拉姆達表示式的匿名函式執行,可以獲取執行返回值
 19 //不支援類成員函式, 支援類靜態成員函式或全域性函式,Opteron()函式等
 20 class threadpool
 21 {
 22     using Task = std::function<void()>;
 23     // 執行緒池
 24     std::vector<std::thread> pool;
 25     // 任務佇列
 26     std::queue<Task> tasks;
 27 
     // 同步
 28     std::mutex m_lock;
 29     // 條件阻塞
 30     std::condition_variable cv_task;
 31     // 是否關閉提交
 32     std::atomic<bool> stoped;
 33     //空閒執行緒數量
 34     std::atomic<int>  idlThrNum;
 35 
 36 public:
 37     inline threadpool(unsigned short size = 4) :stoped{ false }
 38     {
 39         idlThrNum = size < 1 ? 1 : size;
 40         for (size = 0; size < idlThrNum; ++size)
 41         {   //初始化執行緒數量
 42             pool.emplace_back(
 43                 [this]
 44                 { // 工作執行緒函式
 45                     while(!this->stoped)
 46                     {
 47                         std::function<void()> task;
 48                         {   // 獲取一個待執行的 task
 49                             std::unique_lock<std::mutex> lock{ this->m_lock };// unique_lock 相比 lock_guard 的好處是：可以隨時 unlock() 和 lock()
 50                             this->cv_task.wait(lock,
 51                                 [this] {
 52                                     return this->stoped.load() || !this->tasks.empty();
 53                                 }
 54                             ); // wait 直到有 task
 55                             if (this->stoped && this->tasks.empty())
 56                                 return;
 57                             task = std::move(this->tasks.front()); // 取一個 task
 58                             this->tasks.pop();
 59                         }
 60                         idlThrNum--;
 61                         task();
 62                         idlThrNum++;
 63                     }
 64                 }
 65             );
 66         }
 67     }
 68     inline ~threadpool()
 69     {
 70         stoped.store(true);
 71         cv_task.notify_all(); // 喚醒所有執行緒執行
 72         for (std::thread& thread : pool) {
 73             //thread.detach(); // 讓執行緒“自生自滅”
 74             if(thread.joinable())
 75                 thread.join(); // 等待任務結束， 前提：執行緒一定會執行完
 76         }
 77     }
 78 
 79 public:
 80     // 提交一個任務
 81     // 呼叫.get()獲取返回值會等待任務執行完,獲取返回值
 82     // 有兩種方法可以實現呼叫類成員，
 83     // 一種是使用   bind： .commit(std::bind(&Dog::sayHello, &dog));
 84     // 一種是用 mem_fn： .commit(std::mem_fn(&Dog::sayHello), &dog)
 85     template<class F, class... Args>
 86     auto commit(F&& f, Args&&... args) ->std::future<decltype(f(args...))>
 87     {
 88         if (stoped.load())    // stop == true ??
 89             throw std::runtime_error("commit on ThreadPool is stopped.");
 90 
 91         using RetType = decltype(f(args...)); // typename std::result_of<F(Args...)>::type, 函式 f 的返回值型別
 92         auto task = std::make_shared<std::packaged_task<RetType()> >(
 93             std::bind(std::forward<F>(f), std::forward<Args>(args)...)
 94             );    // wtf !
 95         std::future<RetType> future = task->get_future();
 96         {    // 新增任務到佇列
 97             std::lock_guard<std::mutex> lock{ m_lock };//對當前塊的語句加鎖  lock_guard 是 mutex 的 stack 封裝類，構造的時候 lock()，析構的時候 unlock()
 98             tasks.emplace(
 99                 [task]()
100                 { // push(Task{...})
101                     (*task)();
102                 }
103             );
104         }
105         cv_task.notify_one(); // 喚醒一個執行緒執行
106 
107         return future;
108     }
109 
110     //空閒執行緒數量
111     int idlCount() { return idlThrNum; }
112 
113 };
114 
115 }
116 
117 #endif

程式碼不多吧,上百行程式碼就完成了 執行緒池, 並且, 看看 commit, 哈, 不是固定引數的, 無引數數量限制! 這得益於可變引數模板.

怎麼使用?

看下面程式碼(展開檢視)

 1 #include "threadpool.h"
 2 #include <iostream>
 3 
 4 void fun1(int slp)
 5 {
 6     printf("  hello, fun1 !  %d\n" ,std::this_thread::get_id());
 7     if (slp>0) {
 8         printf(" ======= fun1 sleep %d  =========  %d\n",slp, std::this_thread::get_id());
 9         std::this_thread::sleep_for(std::chrono::milliseconds(slp));
10     }
11 }
12 
13 struct gfun {
14     int operator()(int n) {
15         printf("%d  hello, gfun !  %d\n" ,n, std::this_thread::get_id() );
16         return 42;
17     }
18 };
19 
20 class A { 
21 public:
22     static int Afun(int n = 0) {   //函式必須是 static 的才能直接使用執行緒池
23         std::cout << n << "  hello, Afun !  " << std::this_thread::get_id() << std::endl;
24         return n;
25     }
26 
27     static std::string Bfun(int n, std::string str, char c) {
28         std::cout << n << "  hello, Bfun !  "<< str.c_str() <<"  " << (int)c <<"  " << std::this_thread::get_id() << std::endl;
29         return str;
30     }
31 };
32 
33 int main()
34     try {
35         std::threadpool executor{ 50 };
36         A a;
37         std::future<void> ff = executor.commit(fun1,0);
38         std::future<int> fg = executor.commit(gfun{},0);
39         std::future<int> gg = executor.commit(a.Afun, 9999); //IDE提示錯誤,但可以編譯執行
40         std::future<std::string> gh = executor.commit(A::Bfun, 9998,"mult args", 123);
41         std::future<std::string> fh = executor.commit([]()->std::string { std::cout << "hello, fh !  " << std::this_thread::get_id() << std::endl; return "hello,fh ret !"; });
42 
43         std::cout << " =======  sleep ========= " << std::this_thread::get_id() << std::endl;
44         std::this_thread::sleep_for(std::chrono::microseconds(900));
45 
46         for (int i = 0; i < 50; i++) {
47             executor.commit(fun1,i*100 );
48         }
49         std::cout << " =======  commit all ========= " << std::this_thread::get_id()<< " idlsize="<<executor.idlCount() << std::endl;
50 
51         std::cout << " =======  sleep ========= " << std::this_thread::get_id() << std::endl;
52         std::this_thread::sleep_for(std::chrono::seconds(3));
53 
54         ff.get(); //呼叫.get()獲取返回值會等待執行緒執行完,獲取返回值
55         std::cout << fg.get() << "  " << fh.get().c_str()<< "  " << std::this_thread::get_id() << std::endl;
56 
57         std::cout << " =======  sleep ========= " << std::this_thread::get_id() << std::endl;
58         std::this_thread::sleep_for(std::chrono::seconds(3));
59 
60         std::cout << " =======  fun1,55 ========= " << std::this_thread::get_id() << std::endl;
61         executor.commit(fun1,55).get();    //呼叫.get()獲取返回值會等待執行緒執行完
62 
63         std::cout << "end... " << std::this_thread::get_id() << std::endl;
64 
65 
66         std::threadpool pool(4);
67         std::vector< std::future<int> > results;
68 
69         for (int i = 0; i < 8; ++i) {
70             results.emplace_back(
71                 pool.commit([i] {
72                     std::cout << "hello " << i << std::endl;
73                     std::this_thread::sleep_for(std::chrono::seconds(1));
74                     std::cout << "world " << i << std::endl;
75                     return i*i;
76                 })
77             );
78         }
79         std::cout << " =======  commit all2 ========= " << std::this_thread::get_id() << std::endl;
80 
81         for (auto && result : results)
82             std::cout << result.get() << ' ';
83         std::cout << std::endl;
84         return 0;
85     }
86 catch (std::exception& e) {
87     std::cout << "some unhappy happened...  " << std::this_thread::get_id() << e.what() << std::endl;
88 }

View Code

為了避嫌，先進行一下版權說明：程式碼是 me “寫”的，但是思路來自 Internet，特別是這個執行緒池實現(基本 copy 了這個實現,加上這位同學的實現和解釋，好東西值得 copy ! 然後綜合更改了下,更加簡潔)。

實現原理

接著前面的廢話說。“管理一個任務佇列，一個執行緒佇列，然後每次取一個任務分配給一個執行緒去做，迴圈往復。” 這個思路有神馬問題？執行緒池一般要複用執行緒，所以如果是取一個 task 分配給某一個 thread，執行完之後再重新分配，在語言層面基本都是不支援的：一般語言的 thread 都是執行一個固定的 task 函式，執行完畢執行緒也就結束了(至少 c++ 是這樣)。so 要如何實現 task 和 thread 的分配呢？

讓每一個 thread 都去執行排程函式：迴圈獲取一個 task，然後執行之。

idea 是不是很贊！保證了 thread 函式的唯一性，而且複用執行緒執行 task 。

即使理解了 idea，程式碼還是需要詳細解釋一下的。

一個執行緒 pool，一個任務佇列 queue ，應該沒有意見；
任務佇列是典型的生產者-消費者模型，本模型至少需要兩個工具：一個 mutex + 一個條件變數，或是一個 mutex + 一個訊號量。mutex 實際上就是鎖，保證任務的新增和移除(獲取)的互斥性，一個條件變數是保證獲取 task 的同步性：一個 empty 的佇列，執行緒應該等待(阻塞)；
atomic<bool> 本身是原子型別，從名字上就懂：它們的操作 load()/store() 是原子操作，所以不需要再加 mutex。

c++語言細節

即使懂原理也不代表能寫出程式，上面用了眾多c++11的“奇技淫巧”，下面簡單描述之。

using Task = function<void()> 是類型別名，簡化了 typedef 的用法。function<void()> 可以認為是一個函式型別，接受任意原型是 void() 的函式，或是函式物件，或是匿名函式。void() 意思是不帶引數，沒有返回值。
pool.emplace_back([this]{...}) 和 pool.push_back([this]{...}) 功能一樣，只不過前者效能會更好；
pool.emplace_back([this]{...}) 是構造了一個執行緒物件，執行函式是拉姆達匿名函式；
所有物件的初始化方式均採用了 {}，而不再使用 () 方式，因為風格不夠一致且容易出錯；
匿名函式： [this]{...} 不多說。[] 是捕捉器，this 是引用域外的變數 this指標，內部使用死迴圈, 由cv_task.wait(lock,[this]{...}) 來阻塞執行緒；
delctype(expr) 用來推斷 expr 的型別，和 auto 是類似的，相當於型別佔位符，佔據一個型別的位置；auto f(A a, B b) -> decltype(a+b) 是一種用法，不能寫作 decltype(a+b) f(A a, B b)，為啥？！ c++ 就是這麼規定的！
commit 方法是不是略奇葩！可以帶任意多的引數，第一個引數是 f，後面依次是函式 f 的引數！(注意:引數要傳struct/class的話,建議用pointer,小心變數的作用域) 可變引數模板是 c++11 的一大亮點，夠亮！至於為什麼是 Arg... 和 arg... ，因為規定就是這麼用的！
commit 直接使用只能呼叫stdcall函式，但有兩種方法可以實現呼叫類成員，一種是使用 bind： .commit(std::bind(&Dog::sayHello, &dog))；一種是用 mem_fn： .commit(std::mem_fn(&Dog::sayHello), &dog)；
make_shared 用來構造 shared_ptr 智慧指標。用法大體是 shared_ptr<int> p = make_shared<int>(4) 然後 *p == 4 。智慧指標的好處就是，自動 delete ！
bind 函式，接受函式 f 和部分引數，返回currying後的匿名函式，譬如 bind(add, 4) 可以實現類似 add4 的函式！
forward() 函式，類似於 move() 函式，後者是將引數右值化，前者是... 腫麼說呢？大概意思就是：不改變最初傳入的型別的引用型別(左值還是左值，右值還是右值)；
packaged_task 就是任務函式的封裝類，通過 get_future 獲取 future ，然後通過 future 可以獲取函式的返回值(future.get())；packaged_task 本身可以像函式一樣呼叫 () ；
queue 是佇列類， front() 獲取頭部元素， pop() 移除頭部元素；back() 獲取尾部元素，push() 尾部新增元素；
lock_guard 是 mutex 的 stack 封裝類，構造的時候 lock()，析構的時候 unlock()，是 c++ RAII 的 idea；
condition_variable cv; 條件變數，需要配合 unique_lock 使用；unique_lock 相比 lock_guard 的好處是：可以隨時 unlock() 和 lock()。 cv.wait() 之前需要持有 mutex，wait 本身會 unlock() mutex，如果條件滿足則會重新持有 mutex。
最後執行緒池析構的時候,join() 可以等待任務都執行完在結束,很安全!

Git

[copy right from url: http://blog.csdn.net/zdarks/article/details/46994607, https://github.com/progschj/ThreadPool/blob/master/ThreadPool.h]

如果您覺得閱讀本文對您有幫助，請點一下“推薦”按鈕，您的“推薦”將是我最大的寫作動力！歡迎各位轉載，但是未經作者本人同意，轉載文章之後必須在文章頁面明顯位置給出作者和原文連線，否則保留追究法律責任的權利。
原文連結:https://www.cnblogs.com/lzpong/p/6397997.html

基於C++11的執行緒池(threadpool),簡潔且可以帶任意多的引數（轉）

程式碼實現

怎麼使用?

實現原理

c++語言細節

Git

基於C++11的執行緒池(threadpool),簡潔且可以帶任意多的引數（轉）

c++11執行緒池實現

c++11 執行緒池系列之一所需要的join_threads

c++11 執行緒池實現以及示例

Java多執行緒-----執行緒池的使用，原理以及舉例實現（三）（四）：使用樣例及如何配置執行緒池大小

執行緒安全的集合類、CopyOnWrite機制介紹（轉）

基於Linux/C++簡單執行緒池的實現

Java核心（二）深入理解執行緒池ThreadPool

美團面試題：Java-執行緒池 ThreadPool 專題詳解

C++建立執行緒池

C++11執行緒安全佇列和安全棧

c語言執行緒池的簡單實現

c++簡單執行緒池實現

八、JAVA多執行緒：執行緒池原理以及自定義執行緒池 ThreadPool

執行緒組threadgroup、執行緒池threadpool

執行緒池ThreadPool

C++11執行緒安全佇列

[C#]獲得執行緒池中活動的執行緒數

Java-執行緒池 ThreadPool 專題詳解 (美團面試題)

C++簡易執行緒池的實現

基於C++11的執行緒池(threadpool),簡潔且可以帶任意多的引數（轉）

程式碼實現

怎麼使用?

實現原理

c++語言細節

Git

相關推薦