C++ 並發編程之互斥鎖和條件變量的性能比較
來源 https://juejin.im/post/5d6b2e655188257a2615eb72
介紹
本文以最簡單生產者消費者模型，通過運行程序，觀察該進程的cpu使用率，來對比使用互斥鎖 和 互斥鎖+條件變量的性能比較。
本例子的生產者消費者模型，1個生產者，5個消費者。 生產者線程往隊列里放入數據，5個消費者線程從隊列取數據，取數據前需要判斷一下隊列中是否有數據，這個隊列是全局隊列，是線程間共享的數據，所以需要使用互斥鎖進行保護。即生產者在往隊列里放入數據時，其余消費者不能取，反之亦然。
互斥鎖實現的代碼
#include <iostream> // std::cout #include <deque> // std::deque #include <thread> // std::thread #include <chrono> // std::chrono #include <mutex> // std::mutex // 全局隊列 std::deque<int> g_deque; // 全局鎖 std::mutex g_mutex; // 生產者運行標記 bool producer_is_running = true; // 生產者線程函數 void Producer() { // 庫存個數 int count = 8; do { // 智能鎖，初始化后即加鎖，保護的范圍是代碼花括號內，花括號退出即會自動解鎖 // 可以手動解鎖，從而控制互斥鎖的細粒度 std::unique_lock<std::mutex> locker( g_mutex ); // 入隊一個數據 g_deque.push_front( count ); // 提前解鎖，縮小互斥鎖的細粒度，只針對共享的隊列數據進行同步保護 locker.unlock(); std::cout << "生產者 ：我現在庫存有 :" << count << std::endl; // 放慢生產者生產速度，睡1秒 std::this_thread::sleep_for( std::chrono::seconds( 1 ) ); // 庫存自減少 count--; } while( count > 0 ); // 標記生產者打樣了 producer_is_running = false; std::cout << "生產者 ： 我的庫存沒有了，我要打樣了！" << std::endl; } // 消費者線程函數 void Consumer(int id) { int data = 0; do { // 智能鎖，初始化后即加鎖，保護的范圍是代碼花括號內，花括號退出即會自動解鎖 // 可以手動解鎖，從而控制互斥鎖的細粒度 std::unique_lock<std::mutex> locker( g_mutex ); // 隊列不為空 if( !g_deque.empty() ) { // 取出隊列里最后一個數據 data = g_deque.back(); // 刪除隊列里最后一個數據 g_deque.pop_back(); // 提前解鎖，縮小互斥鎖的細粒度，只針對共享的隊列數據進行同步保護 locker.unlock(); std::cout << "消費者[" << id << "] : 我搶到貨的編號是 :" << data << std::endl; } // 隊列為空 else { locker.unlock(); } } while( producer_is_running ); std::cout << "消費者[" << id << "] ：賣家沒有貨打樣了，真可惜，下次再來搶！" << std::endl; } int main(void) { std::cout << "1 producer start ..." << std::endl; std::thread producer( Producer ); std::cout << "5 consumer start ..." << std::endl; std::thread consumer[ 5 ]; for(int i = 0; i < 5; i++) { consumer[i] = std::thread(Consumer, i + 1); } producer.join(); for(int i = 0; i < 5; i++) { consumer[i].join(); } std::cout << "All threads joined." << std::endl; return 0; } 復制代碼互斥鎖實現運行結果：
結果輸出
[root@lincoding condition]# g++ -std=c++0x -pthread -D_GLIBCXX_USE_NANOSLEEP main.cpp -o main
[root@lincoding condition]# ./main
1 producer start ...
5 consumer start ...
生產者 ：我現在庫存有 :8
消費者[1] : 我搶到貨的編號是 :8
消費者[1] : 我搶到貨的編號是 :7
生產者 ：我現在庫存有 :7
生產者 ：我現在庫存有 :6
消費者[3] : 我搶到貨的編號是 :6
生產者 ：我現在庫存有 :5
消費者[1] : 我搶到貨的編號是 :5
生產者 ：我現在庫存有 :4
消費者[2] : 我搶到貨的編號是 :4
生產者 ：我現在庫存有 :3
消費者[5] : 我搶到貨的編號是 :3
生產者 ：我現在庫存有 :2
消費者[2] : 我搶到貨的編號是 :2
生產者 ：我現在庫存有 :1
消費者[1] : 我搶到貨的編號是 :1
生產者 ： 我的庫存沒有了，我要打樣了！消費者[
5] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[2] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[3] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[4] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[1] ：賣家沒有貨打樣了，真可惜，下次再來搶！
All threads joined.
復制代碼可以看到，互斥鎖其實可以完成這個任務，但是卻存在着性能問題。
-
Producer是生產者線程，在生產者數據過程中，會休息1秒，所以這個生產過程是很慢的； -
Consumer是消費者線程，存在着一個while循環，只有判斷到生產者不運行了，才會退出while循環，那么每次在循環體內，都是會先加鎖，判斷隊列不空，然后從列隊取出一個數據，最后解鎖。所以說，在生產者休息1秒的時候，消費者線程實際上會做很多無用功，導致CPU使用率非常高！
運行的環境是4核cpu
[root@lincoding ~]# grep 'model name' /proc/cpuinfo | wc -l
4
復制代碼top命令查看cpu使用情況，可見使用純互斥鎖cpu的開銷是很大的，main進程的cpu使用率達到了357.5%CPU，系統開銷的cpu為54.5%sy，用戶開銷的cpu為18.2%us
[root@lincoding ~]# top
top - 19:13:41 up 36 min, 3 users, load average: 0.06, 0.05, 0.01
Tasks: 179 total, 1 running, 178 sleeping, 0 stopped, 0 zombie
Cpu(s): 18.2%us, 54.5%sy, 0.0%ni, 27.3%id, 0.0%wa, 0.0%hi, 0.0%si, 0.0%st
Mem: 1004412k total, 313492k used, 690920k free, 41424k buffers
Swap: 2031608k total, 0k used, 2031608k free, 79968k cached
PID USER PR NI VIRT RES SHR S %CPU %MEM TIME+ COMMAND
35346 root 20 0 137m 3288 1024 S 357.5 0.3 0:05.92 main
1 root 20 0 19232 1492 1224 S 0.0 0.1 0:02.16 init
2 root 20 0 0 0 0 S 0.0 0.0 0:00.01 kthreadd
3 root RT 0 0 0 0 S 0.0 0.0 0:00.68 migration/0
復制代碼解決的辦法之一就是給消費者也加一個小延時，當消費者沒取到數據時，就休息一下500毫秒，這樣可以減少互斥鎖給cpu帶來的開銷。
// 消費者線程函數 void Consumer(int id) { int data = 0; do { // 智能鎖，初始化后即加鎖，保護的范圍是代碼花括號內，花括號退出即會自動解鎖 // 可以手動解鎖，從而控制互斥鎖的細粒度 std::unique_lock<std::mutex> locker( g_mutex ); // 隊列不為空 if( !g_deque.empty() ) { // 取出隊列里最后一個數據 data = g_deque.back(); // 刪除隊列里最后一個數據 g_deque.pop_back(); // 提前解鎖，縮小互斥鎖的細粒度，只針對共享的隊列數據進行同步保護 locker.unlock(); std::cout << "消費者[" << id << "] : 我搶到貨的編號是 :" << data << std::endl; } // 隊列為空 else { locker.unlock(); // 休息500毫秒 std::this_thread::sleep_for( std::chrono::milliseconds( 500 ) ); } } while( producer_is_running ); std::cout << "消費者[" << id << "] ：賣家沒有貨打樣了，真可惜，下次再來搶！" << std::endl; } 復制代碼從運行結果可知，cpu使用率大大降低了
[root@lincoding ~]# ps aux | grep -v grep |grep main
USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND
root 61296 0.0 0.1 141068 1244 pts/1 Sl+ 19:40 0:00 ./main
復制代碼條件變量+互斥鎖實現的代碼
那么問題來了，如何確定消費者延時（休息）多久呢？
- 如果生產者生產的非常快，消費者卻延時了
500毫秒，也不是很好 - 如果生產者生產的更慢，那么消費延時
500毫秒，也會有無用功，占用了CPU
這就需要引入條件變量std::condition_variable，應用於消費者生產模型中，就是生產者生產完一個數據后，通過notify_one()喚醒正在wait()消費者線程，使得消費者從隊列取出一個數據。
#include <iostream> // std::cout #include <deque> // std::deque #include <thread> // std::thread #include <chrono> // std::chrono #include <mutex> // std::mutex #include <condition_variable> // std::condition_variable // 全局隊列 std::deque<int> g_deque; // 全局鎖 std::mutex g_mutex; // 全局條件變量 std::condition_variable g_cond; // 生產者運行標記 bool producer_is_running = true; // 生產者線程函數 void Producer() { // 庫存個數 int count = 8; do { // 智能鎖，初始化后即加鎖，保護的范圍是代碼花括號內，花括號退出即會自動解鎖 // 可以手動解鎖，從而控制互斥鎖的細粒度 std::unique_lock<std::mutex> locker( g_mutex ); // 入隊一個數據 g_deque.push_front( count ); // 提前解鎖，縮小互斥鎖的細粒度，只針對共享的隊列數據進行同步保護 locker.unlock(); std::cout << "生產者 ：我現在庫存有 :" << count << std::endl; // 喚醒一個線程 g_cond.notify_one(); // 睡1秒 std::this_thread::sleep_for( std::chrono::seconds( 1 ) ); // 庫存自減少 count--; } while( count > 0 ); // 標記生產者打樣了 producer_is_running = false; // 喚醒所有消費線程 g_cond.notify_all(); std::cout << "生產者 ： 我的庫存沒有了，我要打樣了！" << std::endl; } // 消費者線程函數 void Consumer(int id) { // 消費者線程函數 void Consumer(int id) { // 購買的貨品編號 int data = 0; do { // 智能鎖，初始化后即加鎖，保護的范圍是代碼花括號內，花括號退出即會自動解鎖 // 可以手動解鎖，從而控制互斥鎖的細粒度 std::unique_lock<std::mutex> locker( g_mutex ); // wait()函數會先調用互斥鎖的unlock()函數，然后再將自己睡眠，在被喚醒后，又會繼續持有鎖，保護后面的隊列操作 // 必須使用unique_lock，不能使用lock_guard，因為lock_guard沒有lock和unlock接口，而unique_lock則都提供了 g_cond.wait(locker); // 隊列不為空 if( !g_deque.empty() ) { // 取出隊列里最后一個數據 data = g_deque.back(); // 刪除隊列里最后一個數據 g_deque.pop_back(); // 提前解鎖，縮小互斥鎖的細粒度，只針對共享的隊列數據進行同步保護 locker.unlock(); std::cout << "消費者[" << id << "] : 我搶到貨的編號是 :" << data << std::endl; } // 隊列為空 else { locker.unlock(); } } while( producer_is_running ); std::cout << "消費者[" << id << "] ：賣家沒有貨打樣了，真可惜，下次再來搶！" << std::endl; } } int main(void) { std::cout << "1 producer start ..." << std::endl; std::thread producer( Producer ); std::cout << "5 consumer start ..." << std::endl; std::thread consumer[ 5 ]; for(int i = 0; i < 5; i++) { consumer[i] = std::thread(Consumer, i + 1); } producer.join(); for(int i = 0; i < 5; i++) { consumer[i].join(); } std::cout << "All threads joined." << std::endl; return 0; } 復制代碼條件變量+互斥鎖運行結果
[root@lincoding condition]# g++ -std=c++0x -pthread -D_GLIBCXX_USE_NANOSLEEP main.cpp -o main
[root@lincoding condition]#
[root@lincoding condition]# ./main
1 producer start ...
5 consumer start ...
生產者 ：我現在庫存有 :8
消費者[4] : 我搶到貨的編號是 :8
生產者 ：我現在庫存有 :7
消費者[2] : 我搶到貨的編號是 :7
生產者 ：我現在庫存有 :6
消費者[3] : 我搶到貨的編號是 :6
生產者 ：我現在庫存有 :5
消費者[5] : 我搶到貨的編號是 :5
生產者 ：我現在庫存有 :4
消費者[1] : 我搶到貨的編號是 :4
生產者 ：我現在庫存有 :3
消費者[4] : 我搶到貨的編號是 :3
生產者 ：我現在庫存有 :2
消費者[2] : 我搶到貨的編號是 :2
生產者 ：我現在庫存有 :1
消費者[3] : 我搶到貨的編號是 :1
生產者 ： 我的庫存沒有了，我要打樣了！
消費者[5] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[1] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[4] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[2] ：賣家沒有貨打樣了，真可惜，下次再來搶！
消費者[3] ：賣家沒有貨打樣了，真可惜，下次再來搶！
All threads joined.
復制代碼CPU開銷非常的小
[root@lincoding ~]# ps aux | grep -v grep |grep main
USER PID %CPU %MEM VSZ RSS TTY STAT START TIME COMMAND
root 73838 0.0 0.1 141068 1256 pts/1 Sl+ 19:54 0:00 ./main
復制代碼總結
在不確定生產者的生產速度是快還是慢的場景里，不能只使用互斥鎖保護共享的數據，這樣會對CPU的性能開銷非常大，可以使用互斥鎖+條件變量的方式，當生產者線程生產了一個數據，就喚醒消費者線程進行消費，避免一些無用功的性能開銷。
============ End