C/C++ 线程三种并发方式比较（传统互斥量加锁方式, no lock不加锁的方式, 原子函数方式）
執行速度結果：
- 傳統互斥量加鎖方式 <?no lock不加鎖的方式 <?原子函數方式
正文如下:
最近編碼需要實現多線程環境下的計數器操作，統計相關事件的次數。下面是一些學習心得和體會。不敢妄稱原創，基本是學習筆記。遇到相關的引用，我會致謝。
??? 當然我們知道，count++這種操作不是原子的。一個自加操作，本質是分成三步的：
?????1 從緩存取到寄存器
???? 2 在寄存器加1
???? 3 存入緩存。
mov eax,dword ptr [a]
add eax,1
mov dword ptr [a],eax
由于時序的因素，多個線程操作同一個全局變量，會出現問題。這也是并發編程的難點。在目前多核條件下，這種困境會越來越彰顯出來。
最簡單的處理辦法就是加鎖保護，這也是我最初的解決方案?？聪旅娴拇a：
后來在網上查找資料，找到了__sync_fetch_and_add系列的命令
?????__sync_fetch_and_add系列一共有十二個函數，有加/減/與/或/異或/等函數的原子性操作函數,
__snyc_fetch_and_add : 先fetch然后自加,返回的是自加以前的值 __snyc_add_and_fetch : 先自加然后返回,返回的是自加以后的值 (參照 ++i 和 i++)__snyc_fetch_and_add的一個簡單使用 int count = 4; __sync_fetch_and_add(&count, 1); // __sync_fetch_and_add(&count, 1) == 4 cout<<count<<endl; //--->count=5?對于多線程對全局變量進行自加，我們就再也不用理線程鎖了。
下面這行代碼，和上面被pthread_mutex保護的那行代碼作用是一樣的，而且也是線程安全的。
__sync_fetch_and_add( &global_int, 1 );下面是這群函數的全家福，大家看名字就知道是這些函數是干啥的了。
//在用gcc編譯的時候要加上選項 -march=i686
type __sync_fetch_and_add (type *ptr, type value, ...);
type __sync_fetch_and_sub (type *ptr, type value, ...);
type __sync_fetch_and_or (type *ptr, type value, ...);
type __sync_fetch_and_and (type *ptr, type value, ...);
type __sync_fetch_and_xor (type *ptr, type value, ...);
type __sync_fetch_and_nand (type *ptr, type value, ...);
type __sync_add_and_fetch (type *ptr, type value, ...);
type __sync_sub_and_fetch (type *ptr, type value, ...);
type __sync_or_and_fetch (type *ptr, type value, ...);
type __sync_and_and_fetch (type *ptr, type value, ...);
type __sync_xor_and_fetch (type *ptr, type value, ...);
type __sync_nand_and_fetch (type *ptr, type value, ...);
__sync_fetch_and_add,速度是線程鎖的6～7倍
type可以是1,2,3或者8字節長度的int類型,即
int8_t ? ?
uint8_t
int16_t
uint16_t
int32_t
uint32_t
int64_t
uint64_t
后面的可擴展參數(...)用來指出哪些變量需要memory barrier,因為目前gcc實現的是full barrier（類似于linux kernel 中的mb(),表示這個操作之前的所有內存操作不會被重排序到這個操作之后）,所以可以略掉這個參數。
恩.再找個帖子學習學習.http://blog.csdn.net/hzhsan/article/details/25124901
有一個概念叫過無鎖化編程,?知道linux支持的哪些操作是具有原子特性的是理解和設計無鎖化編程算法的基礎
除了上面提到的12個外 還有4個可以實現互斥鎖的功能
//以下兩個函數提供原子的比較和交換, 如果*ptr = oldValue, 就將newValue寫入*ptr //第一個函數在相等并寫入的情況下返回true //第二個函數返回操作之前的值bool __sync_bool_compare_and_swap(type* ptr, type oldValue, type newValue, ....);type __sync_val_compare_and_swap(type* ptr, type oldValue, type newValue, ....);//將*ptr設為value并返回*ptr操作之前的值 type __sync_lock_test_and_set(type *ptr, type value, ....);//置*ptr為0 void __sync_lock_release(type* ptr, ....); __sync_synchronize(...)//作用 : 發出一個full barrier /*關于memory barrier,cpu會對我們的指令進行排序，一般說來會提高程序的效率，但有時候可能造成我們不希望得到的結果，舉一個例子，比如我們有一個硬件設備，它有4個寄存器，當你發出一個操作指令的時候，一個寄存器存的是你的操作指令（比如READ），兩個寄存器存的是參數（比如是地址和size），最后一個寄存器是控制寄存器，在所有的參數都設置好之后向其發出指令，設備開始讀取參數，執行命令，程序可能如下：*/ write1(dev.register_size, size); write1(dev.register_addr, addr); write1(dev.register_cmd, Read); write1(dev.register_control, GO); /*如果最后一條write1被換到了前幾條語句之前，那么肯定不是我們所期望的，這時候我們可以在最后一條語句之前加入一個memory barrier,強制cpu執行完前面的寫入以后再執行最后一條：*/ write1(dev.register_size, size); write1(dev.register_addr, addr); write1(dev.register_cmd, Read); __sync_synchronize(); write1(dev.register_control, GO);//memory barrier有幾種類型： //acquire barrier : 不允許將barrier之后的內存讀取指令移到barrier之前（linux kernel中的wmb()） //release barrier : 不允許將barrier之前的內存讀取指令移到barrier之后 (linux kernel中的rmb()) //full barrier : 以上兩種barrier的合集(linux kernel中的mb())//好吧,說實話這個函數的說明基本沒看懂最后從網上找一個代碼寫一寫:http://blog.csdn.net/hzhsan/article/details/25837189
測試場景：假設有一個應用：現在有一個全局變量，用來計數，再創建10個線程并發執行，每個線程中循環對這個全局變量進行++操作（i++)，循環加2000000次。
所以很容易知道，這必然會涉及到并發互斥操作。下面通過三種方式[傳統互斥量加鎖方式, no lock不加鎖的方式, 原子函數方式]來實現這種并發操作。并對比出其在效率上的不同之處。
這里先貼上代碼，共5個文件：2個用于做時間統計的文件：timer.h? timer.cpp。這兩個文件是臨時封裝的，只用來計時，可以不必細看。
//timer.h 用于計時#ifndef TIMER_H_ #define TIMER_H_#include <sys/time.h>class Timer {public:Timer();Timer(const Timer& t) = delete;~Timer();void start();void stop();void reset();double costTime();private: struct timeval t1;struct timeval t2;bool b1, b2; }; #endif //timer.cpp #include "timer.h" #include <iostream>using namespace std;Timer::Timer():b1(false), b2(false) { } Timer::~Timer() { } void Timer::start() {gettimeofday(&t1, NULL);b1 = true;b2 = false; } void Timer::stop() {gettimeofday(&t2, NULL);b2 = true; } void Timer::reset() {b1 = false;b2 = false; } double Timer::costTime() {if (!b1){cout<<"error, do not call function start()"<<endl;cout<<"the right sequence : start() ..... stop() costTime()"<<endl;return 0;}if (!b2){cout<<"error, do not call function stop()"<<endl;cout<<"the right sequence : start() ..... stop() costTime()"<<endl;return 0;}size_t sec = t2.tv_sec - t1.tv_sec; double usec = t2.tv_usec - t1.tv_usec;if (sec < 0) {cout<<"error, call stop() before start()"<<endl;cout<<"the right sequence : start() ..... stop() costTime()"<<endl;return 0;}if (usec < 0){usec += 1000000;--sec;if (sec < 0) {cout<<"error, call stop() before start()"<<endl;cout<<"the right sequence : start() ..... stop() costTime()"<<endl;return 0;}}return sec + usec * 1.0 / 1000000; } //thread_function.h -->多線程要調用的函數 #ifndef THREAD_FUNCTION_H_ #define THREAD_FUNCTION_H_ void* thread_lock_execFunc(void* arg); void* thread_nolock_execFunc(void* arg); void* thread_atom_execFunc(void* arg); #endif //thread_function.cpp #include "thread_function.h" #include "lock.h" #include <pthread.h> #include <unistd.h>extern volatile int count; struct LOCK;void* thread_lock_execFunc(void* arg) {for (int i = 0; i < 2000000; ++i){pthread_mutex_lock(reinterpret_cast<pthread_mutex_t*>(arg));++count;pthread_mutex_unlock(reinterpret_cast<pthread_mutex_t*>(arg));}return NULL; }void* thread_nolock_execFunc(void* arg) {LOCK* pLock = reinterpret_cast<LOCK*>(arg);for (int i = 0; i < 2000000; ++i){while(!(__sync_bool_compare_and_swap(&(pLock->mutex), pLock->use, 1))){usleep(100000);}++count;__sync_bool_compare_and_swap(&(pLock->mutex), pLock->unUse, 0);}return NULL; }void* thread_atom_execFunc(void* arg) {for (int i = 0; i < 2000000; ++i){__sync_fetch_and_add(&count, 1);}return NULL; } //lock.h --->給mainnolock.cpp使用的類 #ifndef LOCK_H_ #define LOCK_H_ struct LOCK {int mutex;int use;int unUse;LOCK() : mutex(0), use(0), unUse(1){} }; #endif //mainlock.cpp 使用mutex加鎖方式的多線程 #include <iostream> #include <pthread.h> #include <iomanip>#include "timer.h" #include "thread_function.h"using namespace std;pthread_mutex_t mutex_lock; volatile int count = 0;int main( int argc, char** argv) {pthread_mutex_init(&mutex_lock, NULL);Timer timer;timer.start();/*test thread begin*/pthread_t thread_ids[10];for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i){pthread_create(&thread_ids[i], NULL, thread_lock_execFunc, &mutex_lock);}for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i){pthread_join(thread_ids[i], NULL);}/*test thread end*/timer.stop();cout<<setiosflags(ios::fixed)<<setprecision(4)<<"lock cost["<<timer.costTime()<<"]second"<<endl;return 0; } //main_nolock.cpp 使用__sync_compare_and_swap的多線程 #include <iostream> #include <pthread.h> #include <unistd.h> #include <iomanip> #include "timer.h" #include "thread_function.h" #include "lock.h"using namespace std;volatile int count = 0;int main(int argc, char** argv) {LOCK lock;Timer timer;timer.start();/*test thread begin*/pthread_t thread_ids[10];for (int i = 0; i < sizeof(thread_ids) / sizeof(pthread_t); ++i){pthread_create(&thread_ids[i], NULL, thread_nolock_execFunc, &lock);}for (int i = 0; i < sizeof(thread_ids) / sizeof(pthread_t); ++i){pthread_join(thread_ids[i], NULL);}/*test thread end*/timer.stop();cout<<setiosflags(ios::fixed)<<setprecision(4)<<"nolock cost["<<timer.costTime()<<"]\n"; return 0; } //main_atomic.cpp 使用__sync_fetch_and_add的多線程 #include<iostream> #include<pthread.h> #include<unistd.h> #include<iomanip> #include "timer.h" #include "thread_function.h"using namespace std;volatile int count = 0;int main(int argc, char** argv) {Timer timer;timer.start();/*pthread begin*/pthread_t thread_ids[10];for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i){pthread_create(&thread_ids[i], NULL, thread_atom_execFunc, NULL);}for (int i = 0; i < sizeof(thread_ids)/sizeof(pthread_t); ++i){pthread_join(thread_ids[i], NULL);}/*pthread end*/timer.stop();cout<<setiosflags(ios::fixed)<<setprecision(4)<<"atomic cost["<<timer.costTime()<<"]\n";return 0; } //makefileCC = g++ CFLAGS = -g -lpthread -std=c++11OBJS_LOCK = main_lock.o timer.o thread_function.o OBJS_UNLOCK = main_nolock.o timer.o thread_function.o OBJS_ATOMICLOCK = main_atomic.o timer.o thread_function.oINC = timer.h thread_function.h lock.hlock : $(OBJS_LOCK) $(INC) $(CC) -o mainlock $(OBJS_LOCK) $(CFLAGS)rm *.onolock : $(OBJS_UNLOCK) $(INC)$(CC) -o mainnolock $(OBJS_UNLOCK) $(CFLAGS)rm *.oatomiclock : $(OBJS_ATOMICLOCK) $(INC)$(CC) -o mainatomic $(OBJS_ATOMICLOCK) $(CFLAGS)main_lock.o : main_lock.cpp $(CC) -c main_lock.cpp $(CFLAGS)main_nolock.o : main_nolock.cpp $(CC) -c main_nolock.cpp $(CFLAGS)main_atomic.o : main_atomic.cpp$(CC) -c main_atomic.cpp $(CFLAGS)timer.o : timer.cpp $(CC) -c timer.cpp $(CFLAGS)thread_function.o : thread_function.cpp$(CC) -c thread_function.cpp $(CFLAGS)clean:rm *.o執行makefile
make lock
make nolock
make atomiclock
然后生成3個可執行文件
運行這3個可執行文件:
另外:針對main_nolock.cpp而言,作者提到了一個現象
在thread_function.cpp中, 隨著一下代碼的改變,運行時間會有變化
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ));?
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) )) usleep(1);
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(10);
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(100);
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(1000);
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(10000);
while (!(__sync_bool_compare_and_swap (&mutex,lock, 1) ))usleep(100000);
執行時間的關系是 ?: ? ?T(;)<T(1)<T(10)<T(100)<T(1000)<T(10000)>T(100000)
?通過編程測試及測試得出結論：
1、如果是想用全局變量來做統計操作。而又不得不考慮多線程間的互斥訪問的話，最好使用編譯器支持的原子操作函數。再滿足互斥訪問的前提下，編程最簡單，效率最高。
2、lock-free，無鎖編程方式確實能夠比傳統加鎖方式效率高。所以在高并發程序中采用無鎖編程的方式可以進一步提高程序效率。但是得對無鎖方式有足夠熟悉的了解，不然效率反而會更低而且容易出錯。(比如在某些情況下main_nolock比main_lock的效率還要低)
在學習一個無鎖化編程的分析帖子?http://blog.csdn.net/hzhsan/article/details/25141421
Lock-free 算法通常比基于鎖的算法要好：
- 從其定義來看，它們是 wait-free 的，可以確保線程永遠不會阻塞。
- 狀態轉變是原子性的，以至于在任何點失敗都不會惡化數據結構。
- 因為線程永遠不會阻塞，所以當同步的細粒度是單一原子寫或比較交換時，它們通?？梢詭砀叩耐掏铝?/strong>。
- 在某些情況下，lock-free 算法會有更少的同步寫操作（比如 Interlocked 操作），因此純粹從性能來看，它可能更便宜。
但是 lock-freedom 并不是萬能藥。下面是一些很明顯的不利因素：
- 樂觀的并發使用會對 hot data structures 導致 livelock。
- 代碼需要大量困難的測試。通常其正確性取決于對目標機器內存模型的正確解釋。
- 基于眾多原因，lock-free 代碼很難編寫和維護。
無鎖編程主要是使用原子操作替代鎖來實現對共享資源的訪問保護，舉個例子，要對某個整數變量進行加1操作的話，用鎖保護操作的代碼如下：
int a = 0;
Lock();
a+= 1;
Unlock();
如果對上述代碼反編譯可以發現 a+=1;被翻譯成了以下三條匯編指令：
mov eax,dword ptr [a]
add eax,1
mov dword ptr [a],eax
如果在單核系統中,由于在上述三條指令的任何一條執行完后都可能發生任務切換，比如執行完第1條指令后就發生了任務切換，這時如果有其他任務來對a進行操作的話，當任務切換回來后，將繼續對a進行操作，很可能出現不可預測的結果，因此上述三條指令必須使用鎖來保護，以使這段時間內其他任務無法對a進行操作。
需要注意的是，在多核系統中，因為多個CPU核在物理上是并行的，可能發生同時寫的現象；所以必須保證一個CPU核在對共享內存進行寫操作時，其他CPU核不能寫這塊內存。因此在多核系統中和單核有區別，即使只有一條指令，也需要要加鎖保護。
如果使用原子操作來實現上述加1操作的話，例如使用VC里的InterlockedIncrement來操作的話，那么對a的加1操作需要以下語句
InterlockedIncrement (&a);
這條語句最終的實際加1操作會被翻譯成以下一條帶lock前綴的匯編指令：
lock xadd dword ptr [ecx],eax
使用原子操作時，在進行實際的寫操作時，使用了lock指令，這樣就可以阻止其他任務寫這塊內存，避免出現數據競爭現象。原子操作速度比鎖快，一般要快一倍以上。
使用lock前綴的指令實際上在系統中是使用了內存柵障（memory barrier），當原子操作在進行時，其他任務都不能對內存操作，會影響其他任務的執行。因此這種原子操作實際上屬于一種激烈競爭的鎖，不過由于它的操作時間很快，因此可以看成是一種極細粒度鎖。
在無鎖(Lock－free)編程環境中，主要使用的原子操作為CAS（Compare and Swap）操作，在VC里對應的操作為InterlockedCompareExchange或者InterlockedCompareExchangeAcquire；如果是64位的操作，需要使用InterlockedCompareExchange64或者InterlockedCompareExchangeAcquire64。使用這種原子操作替代鎖的最大的一個好處是它是非阻塞的。
| ? | 比較項目 | 無鎖編程 | 分布式編程 |
| 1 | 加速比性能 | 取決于競爭方式，除非也采用分布式競爭，否則不如分布式鎖競爭的性能 | 加速比和CPU核數成正比關系，接近于單核多任務時的性能 |
| 2 | 實現的功能 | 有限 | 不受限制 |
| 3 | 程序員掌握難易程度 | 難度太高，過于復雜，普通程序員無法掌握，目前世界上只有少數幾個人掌握。 | 和單核時代的數據結構算法難度差不多，普通程序員可以掌握 |
| 4 | 現有軟件的移植 | 使用無鎖算法后，以往的算法需要廢棄掉，無法復用 | 可以繼承已有的算法，在已有程序基礎上重構即可。 |
從上表的四個方面的綜合比較可以看出，無鎖編程的實用價值是遠遠不如分布式編程的，因此分布式編程比無鎖編程更適合多核CPU系統。
可在分布計算機系統的幾臺計算機上同時協調執行的程序設計方法,分布式程序設計的主要特征是分布和通信。采用分布式程序設計方法設計程序時，一個程序由若干個可獨立執行的程序模塊組成。這些程序模塊分布于一個分布式計算機系統的幾臺計算機上同時執行。分布在各臺計算機上的程序模塊是相互關聯的，它們在執行中需要交換數據，即通信。只有通過通信，各程序模塊才能協調地完成一個共同的計算任務。采用分布式程序設計方法解決計算問題時，必須提供用以進行分布式程序設計的語言和設計相應的分布式算法。分布式程序設計語言與常用的各種程序設計語言的主要區別，在于它具有程序分布和通信的功能。因此，分布式程序設計語言，往往可以由一種程序設計語言增加分布和通信的功能而構成。分布式算法和適用于多處理器系統的并行算法，都具有并行執行的特點，但它們是有區別的。設計分布式算法時，必須保證實現算法的各程序模塊間不會有公共變量，它們只能通過通信來交換數據。此外，設計分布式算法時，往往需要考慮堅定性，即當系統中幾臺計算機失效時，算法仍是有效的。
總結
以上是生活随笔為你收集整理的C/C++ 线程三种并发方式比较（传统互斥量加锁方式, no lock不加锁的方式, 原子函数方式）的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： C/C++ 去掉字符串首位的空格字符
- 下一篇： C/C++取绝对值