多線程不進(jìn)行同步會造成什么問題

背景問題：在特定的應(yīng)用場景下，多線程不進(jìn)行同步會造成什么問題？

通過多線程模擬多窗口售票為例：

#include

#include

#include

#include

#include

#include

using namespace std;

int ticket_sum=20;

void *sell_ticket(void *arg)

{

for(int i=0; i<20; i++)

{

if(ticket_sum>0)

{

sleep(1);

cout<<"sell the "<<20-ticket_sum+1<<"th"<

ticket_sum--;

}

}

return 0;

}

int main()

{

int flag;

pthread_t tids[4];

for(int i=0; i<4; i++)

{

flag=pthread_create(&tids[i],NULL,&sell_ticket,NULL);

if(flag)

{

cout<<"pthread create error ,flag="<

return flag;

}

}

sleep(20);

void *ans;

for(int i=0; i<4; i++)

{

flag=pthread_join(tids[i],&ans);

if(flag)

{

cout<<"tid="<

分析：總票數(shù)只有20張，卻賣出了23張，是非常明顯的超買超賣問題，而造成這個問題的根本原因就是同時發(fā)生的各個線程都可以對ticket_sum進(jìn)行讀取和寫入！

ps：

1.在并發(fā)情況下，指令執(zhí)行的先后順序由內(nèi)核決定，同一個線程內(nèi)部，指令按照先后順序執(zhí)行，但不同線程之間的指令很難說清楚是哪一個先執(zhí)行，如果運(yùn)行的結(jié)果依賴于不同線程執(zhí)行的先后的話，那么就會形成競爭條件，在這樣的情況下，計算的結(jié)果很難預(yù)知，所以應(yīng)該盡量避免競爭條件的形成

2.最常見的解決競爭條件的方法是將原先分離的兩個指令構(gòu)成一個不可分割的原子操作，而其他任務(wù)不能插入到原子操作中！

3.對多線程來說，同步指的是在一定時間內(nèi)只允許某一個線程訪問某個資源，而在此時間內(nèi)，不允許其他線程訪問該資源!

4.線程同步的常見方法：互斥鎖，條件變量，讀寫鎖，信號量

一.互斥鎖

本質(zhì)就是一個特殊的全局變量，擁有l(wèi)ock和unlock兩種狀態(tài)，unlock的互斥鎖可以由某個線程獲得，一旦獲得，這個互斥鎖會鎖上變成lock狀態(tài)，此后只有該線程由權(quán)力打開該鎖，其他線程想要獲得互斥鎖，必須得到互斥鎖再次被打開之后

采用互斥鎖來同步資源:

#include

#include

#include

#include

#include

#include

using namespace std;

int ticket_sum=20;

pthread_mutex_t mutex_x=PTHREAD_MUTEX_INITIALIZER;//static init mutex

void *sell_ticket(void *arg)

{

for(int i=0; i<20; i++)

{

pthread_mutex_lock(&mutex_x);//atomic opreation through mutex lock

if(ticket_sum>0)

{

sleep(1);

cout<<"sell the "<<20-ticket_sum+1<<"th"<

ticket_sum--;

}

pthread_mutex_unlock(&mutex_x);

}

return 0;

}

int main()

{

int flag;

pthread_t tids[4];

for(int i=0; i<4; i++)

{

flag=pthread_create(&tids[i],NULL,&sell_ticket,NULL);

if(flag)

{

cout<<"pthread create error ,flag="<

return flag;

}

}

sleep(20);

void *ans;

for(int i=0; i<4; i++)

{

flag=pthread_join(tids[i],&ans);

if(flag)

{

cout<<"tid="<

ticket_sum--;

}

sleep(1);

pthread_mutex_unlock(&mutex_x);

sleep(1);

}

return 0;

}

void *sell_ticket_2(void *arg)

{

int flag;

for(int i=0; i<10; i++)

{

flag=pthread_mutex_trylock(&mutex_x);

if(flag==EBUSY)

{

cout<<"sell_ticket_2:the variable is locked by sell_ticket_1"<

}

else if(flag==0)

{

if(ticket_sum>0)

{

sleep(1);

cout<<"thread_2 sell the "<<20-ticket_sum+1<<"th tickets"<

ticket_sum--;

}

pthread_mutex_unlock(&mutex_x);

}

sleep(1);

}

return 0;

}

int main()

{

int flag;

pthread_t tids[2];

flag=pthread_create(&tids[0],NULL,&sell_ticket_1,NULL);

if(flag)

{

cout<<"pthread create error ,flag="<

return flag;

}

flag=pthread_create(&tids[1],NULL,&sell_ticket_2,NULL);

if(flag)

{

cout<<"pthread create error ,flag="<

return flag;

}

void *ans;

sleep(30);

flag=pthread_join(tids[0],&ans);

if(flag)

{

cout<<"tid="<

分析：通過測試加鎖函數(shù)我們可以清晰的看到兩個線程爭用資源的情況

二.條件變量

互斥量不是萬能的，比如某個線程正在等待共享數(shù)據(jù)內(nèi)某個條件出現(xiàn)，可可能需要重復(fù)對數(shù)據(jù)對象加鎖和解鎖（輪詢），但是這樣輪詢非常耗費(fèi)時間和資源，而且效率非常低，所以互斥鎖不太適合這種情況

我們需要這樣一種方法：當(dāng)線程在等待滿足某些條件時使線程進(jìn)入睡眠狀態(tài)，一旦條件滿足，就換線因等待滿足特定條件而睡眠的線程

如果我們能夠?qū)崿F(xiàn)這樣一種方法，程序的效率無疑會大大提高，而這種方法正是條件變量！

樣例：

#include

#include

#include

#include

#include

#include

#include

using namespace std;

pthread_cond_t qready=PTHREAD_COND_INITIALIZER; //cond

pthread_mutex_t qlock=PTHREAD_MUTEX_INITIALIZER; //mutex

int x=10,y=20;

void *f1(void *arg)

{

cout<<"f1 start"<

pthread_mutex_lock(&qlock);

while(x

{

pthread_cond_wait(&qready,&qlock);

}

pthread_mutex_unlock(&qlock);

sleep(3);

cout<<"f1 end"<

return 0;

}

void *f2(void *arg)

{

cout<<"f2 start"<

pthread_mutex_lock(&qlock);

x=20;

y=10;

cout<<"has a change,x="<

return 0;

}

int main()

{

pthread_t tids[2];

int flag;

flag=pthread_create(&tids[0],NULL,f1,NULL);

if(flag)

{

cout<<"pthread 1 create error "<

return flag;

}

sleep(2);

flag=pthread_create(&tids[1],NULL,f2,NULL);

if(flag)

{

cout<<"pthread 2 create erro "<

return flag;

}

sleep(5);

return 0;

}

分析：線程1不滿足條件被阻塞，然后線程2運(yùn)行，改變了條件，線程2發(fā)行條件改變了通知線程1運(yùn)行，然線程1不滿足條件被阻塞，然后線程2運(yùn)行，改變了條件，線程2發(fā)行條件改變了通知線程1運(yùn)行，然后線程2結(jié)束，然后線程1繼續(xù)運(yùn)行，然后線程1結(jié)束，為了確保線程1先執(zhí)行，在創(chuàng)建線程2之前我們sleep了2秒

ps：

1.條件變量通過運(yùn)行線程阻塞和等待另一個線程發(fā)送信號的方法彌補(bǔ)互斥鎖的不足，常常和互斥鎖一起使用，使用時，條件變量被用來阻塞一個線程，當(dāng)條件不滿足時，線程往往解開響應(yīng)的互斥鎖并等待條件發(fā)生變化，一旦其他的某個線程改變了條件變量，它將通知響應(yīng)的條件變量換線一個或多個正被此條件變量阻塞的線程，這些線程將重新鎖定互斥鎖并且重新測試條件是否滿足

1.條件變量的相關(guān)函數(shù)

1）創(chuàng)建

靜態(tài)方式：pthread_cond_t cond PTHREAD_COND_INITIALIZER

動態(tài)方式：int pthread_cond_init(&cond,NULL)

Linux thread 實現(xiàn)的條件變量不支持屬性，所以NULL(cond_attr參數(shù)）

2）注銷

int pthread_cond_destory(&cond)

只有沒有線程在該條件變量上，該條件變量才能注銷，否則返回EBUSY

因為Linux實現(xiàn)的條件變量沒有分配什么資源，所以注銷動作只包括檢查是否有等待線程!(請參考條件變量的底層實現(xiàn)）

3）等待

條件等待：int pthread_cond_wait(&cond,&mutex)

計時等待：int pthread_cond_timewait(&cond,&mutex,time)

1.其中計時等待如果在給定時刻前條件沒有被滿足，則返回ETIMEOUT，結(jié)束等待

2.無論那種等待方式，都必須有一個互斥鎖配合，以防止多個線程同時請求pthread_cond_wait形成競爭條件！

3.在調(diào)用pthread_cond_wait前必須由本線程加鎖

4）激發(fā)

激發(fā)一個等待線程：pthread_cond_signal(&cond)

激發(fā)所有等待線程：pthread_cond_broadcast(&cond)

重要的是，pthread_cond_signal不會存在驚群效應(yīng)，也就是是它最多給一個等待線程發(fā)信號，不會給所有線程發(fā)信號喚醒提他們，然后要求他們自己去爭搶資源！

pthread_cond_signal會根據(jù)等待線程的優(yōu)先級和等待時間來確定激發(fā)哪一個等待線程

下面看一個程序，找到程序存在的問題

#include

#include

#include

#include

#include

#include

#include

using namespace std;

pthread_cond_t taxi_cond=PTHREAD_COND_INITIALIZER; //taix arrive cond

pthread_mutex_t taxi_mutex=PTHREAD_MUTEX_INITIALIZER;// sync mutex

void *traveler_arrive(void *name)

{

cout<<"Traveler:"<<(char*)name<<" needs a taxi now!"<

pthread_mutex_lock(&taxi_mutex);

pthread_cond_wait(&taxi_cond,&taxi_mutex);

pthread_mutex_unlock(&taxi_mutex);

cout<<"Traveler:"<<(char*)name<<" now got a taxi!"<

pthread_exit((void*)0);

}

void *taxi_arrive(void *name)

{

cout<<"Taxi:"<<(char*)name<<" arriver."<

pthread_cond_signal(&taxi_cond);

pthread_exit((void*)0);

}

int main()

{

pthread_t tids[3];

int flag;

flag=pthread_create(&tids[0],NULL,taxi_arrive,(void*)("Jack"));

if(flag)

{

cout<<"pthread_create error:flag="<

return flag;

}

cout<<"time passing by"<

sleep(1);

flag=pthread_create(&tids[1],NULL,traveler_arrive,(void*)("Susan"));

if(flag)

{

cout<<"pthread_create error:flag="<

return flag;

}

cout<<"time passing by"<

sleep(1);

flag=pthread_create(&tids[2],NULL,taxi_arrive,(void*)("Mike"));

if(flag)

{

cout<<"pthread_create error:flag="<

return flag;

}

cout<<"time passing by"<

sleep(1);

void *ans;

for(int i=0; i<3; i++)

{

flag=pthread_join(tids[i],&ans);

if(flag)

{

cout<<"pthread_join error:flag="<

return flag;

}

cout<<"ans="<

}

return 0;

}

分析：程序由一個條件變量，用于提示乘客有出租車到達(dá)，還有一個同步鎖，乘客到達(dá)之后就是等車（條件變量），出租車到達(dá)之后就是通知乘客，我們看到乘客Susan到達(dá)之后，并沒有乘坐先到的Jack的車，而是等到Mike的車到了之后再乘坐Mike的車，Jack的車白白的閑置了，為什么會造成這種原因呢？分析一下代碼：我們發(fā)現(xiàn)Jack出租車到達(dá)之后調(diào)用pthread_cond_signal(&taxi_cond)發(fā)現(xiàn)沒有乘客，然后就直接結(jié)束線程了。。。。

正確的操作應(yīng)該是：先到的Jack發(fā)現(xiàn)沒有乘客，然后一直等待乘客，有乘客到了就直接走，而且我們應(yīng)該統(tǒng)計一下乘客的數(shù)量

做如下改進(jìn)：

1.增加乘客計數(shù)器，使得出租車在有乘客到達(dá)之后可以直接走，而不是又在原地等待別的乘客（僵死線程）

2.出租車到達(dá)函數(shù)加個while循環(huán)，沒有乘客的時候一直等待，直到乘客到來

#include

#include

#include

#include

#include

#include

#include

using namespace std;

pthread_cond_t taxi_cond=PTHREAD_COND_INITIALIZER; //taix arrive cond

pthread_mutex_t taxi_mutex=PTHREAD_MUTEX_INITIALIZER;// sync mutex

void *traveler_arrive(void *name)

{

cout<<"Traveler:"<<(char*)name<<" needs a taxi now!"<

pthread_mutex_lock(&taxi_mutex);

pthread_cond_wait(&taxi_cond,&taxi_mutex);

pthread_mutex_unlock(&taxi_mutex);

cout<<"Traveler:"<<(char*)name<<" now got a taxi!"<

pthread_exit((void*)0);

}

void *taxi_arrive(void *name)

{

cout<<"Taxi:"<<(char*)name<<" arriver."<

pthread_exit((void*)0);

}

int main()

{

pthread_t tids[3];

int flag;

flag=pthread_create(&tids[0],NULL,taxi_arrive,(void*)("Jack"));

if(flag)

{

cout<<"pthread_create error:flag="<

return flag;

}

cout<<"time passing by"<

sleep(1);

flag=pthread_create(&tids[1],NULL,traveler_arrive,(void*)("Susan"));

if(flag)

{

cout<<"pthread_create error:flag="<

return flag;

}

cout<<"time passing by"<

sleep(1);

flag=pthread_create(&tids[2],NULL,taxi_arrive,(void*)("Mike"));

if(flag)

{

cout<<"pthread_create error:flag="<

return flag;

}

cout<<"time passing by"<

sleep(1);

void *ans;

for(int i=0; i<3; i++)

{

flag=pthread_join(tids[i],&ans);

if(flag)

{

cout<<"pthread_join error:flag="<

return flag;

}

cout<<"ans="<

}

return 0;

}

三.讀寫鎖

可以多個線程同時讀，但是不能多個線程同時寫

1.讀寫鎖比互斥鎖更加具有適用性和并行性

2.讀寫鎖最適用于對數(shù)據(jù)結(jié)構(gòu)的讀操作讀操作次數(shù)多余寫操作次數(shù)的場合！

3.鎖處于讀模式時可以線程共享，而鎖處于寫模式時只能獨(dú)占，所以讀寫鎖又叫做共享-獨(dú)占鎖

4.讀寫鎖有兩種策略：強(qiáng)讀同步和強(qiáng)寫同步

在強(qiáng)讀同步中，總是給讀者更高的優(yōu)先權(quán)，只要寫者沒有進(jìn)行寫操作，讀者就可以獲得訪問權(quán)限

在強(qiáng)寫同步中，總是給寫者更高的優(yōu)先權(quán)，讀者只能等到所有正在等待或者執(zhí)行的寫者完成后才能進(jìn)行讀

不同的系統(tǒng)采用不同的策略，比如航班訂票系統(tǒng)使用強(qiáng)寫同步，圖書館查閱系統(tǒng)采用強(qiáng)讀同步

根據(jù)不同的業(yè)務(wù)場景，采用不同的策略

1）初始化的銷毀讀寫鎖

靜態(tài)初始化：pthread_rwlock_t rwlock=
PTHREAD_RWLOCK_INITIALIZER

動態(tài)初始化：int pthread_rwlock_init(rwlock，NULL)，NULL代表讀寫鎖采用默認(rèn)屬性

銷毀讀寫鎖：int pthread_rwlock_destory(rwlock)

在釋放某個讀寫鎖的資源之前，需要先通過pthread_rwlock_destory函數(shù)對讀寫鎖進(jìn)行清理。釋放由pthread_rwlock_init函數(shù)分配的資源

如果你想要讀寫鎖使用非默認(rèn)屬性，則attr不能為NULL，得給attr賦值

int pthread_rwlockattr_init(attr),給attr初始化

int
pthread_rwlockattr_destory(attr)，銷毀attr

2）以寫的方式獲取鎖，以讀的方式獲取鎖，釋放讀寫鎖

int pthread_rwlock_rdlock(rwlock),以讀的方式獲取鎖

int pthread_rwlock_wrlock(rwlock)，以寫的方式獲取鎖

int pthread_rwlock_unlock(rwlock),釋放鎖

上面兩個獲取鎖的方式都是阻塞的函數(shù)，也就是說獲取不到鎖的話，調(diào)用線程不是立即返回，而是阻塞執(zhí)行，在需要進(jìn)行寫操作的時候，這種阻塞式獲取鎖的方式是非常不好的，你想一下，我需要進(jìn)行寫操作，不但沒有獲取到鎖，我還一直在這里等待，大大拖累效率

所以我們應(yīng)該采用非阻塞的方式獲取鎖：

int pthread_rwlock_tryrdlock(rwlock)

int pthread_rwlock_trywrlock(rwlock)

讀寫鎖的樣例：

#include

#include

#include

#include

#include

#include

#include

using namespace std;

int num=5;

pthread_rwlock_t rwlock;

void *reader(void *arg)

{

pthread_rwlock_rdlock(&rwlock);

cout<<"reader "<<(long)arg<<" got the lock"<

pthread_rwlock_unlock(&rwlock);

return 0;

}

void *writer(void *arg)

{

pthread_rwlock_wrlock(&rwlock);

cout<<"writer "<<(long)arg<<" got the lock"<

pthread_rwlock_unlock(&rwlock);

return 0;

}

int main()

{

int flag;

long n=1,m=1;

pthread_t wid,rid;

pthread_attr_t attr;

flag=pthread_rwlock_init(&rwlock,NULL);

if(flag)

{

cout<<"rwlock init error"<

return flag;

}

pthread_attr_init(&attr);

pthread_attr_setdetachstate(&attr,PTHREAD_CREATE_DETACHED);//thread sepatate

for(int i=0;i

{

if(i%3)

{

pthread_create(&rid,&attr,reader,(void *)n);

cout<<"create reader "<

n++;

}else

{

pthread_create(&wid,&attr,writer,(void *)m);

cout<<"create writer "<

m++;

}

}

sleep(5);//wait other done

return 0;

}

分析：3個讀線程，2個寫線程，讀線程比寫線程多

當(dāng)讀寫鎖是寫狀態(tài)時，在鎖被解鎖之前，所有試圖對這個鎖加鎖的線程都會被阻塞

當(dāng)讀寫鎖是讀狀態(tài)時，在鎖被解鎖之前，所有視圖以讀模式對它進(jìn)行加鎖的線程都可以得到訪問權(quán)，但是以寫模式對它進(jìn)行加鎖的線程會被阻塞

所以讀寫鎖默認(rèn)是強(qiáng)讀模式！

四.信號量

信號量（sem）和互斥鎖的區(qū)別：互斥鎖只允許一個線程進(jìn)入臨界區(qū)，而信號量允許多個線程進(jìn)入臨界區(qū)

1）信號量初始化

int sem_init(&sem,pshared,v)

pshared為0表示這個信號量是當(dāng)前進(jìn)程的局部信號量

pshared為1表示這個信號量可以在多個進(jìn)程之間共享

v為信號量的初始值

成功返回0，失敗返回-1

2）信號量值的加減

int sem_wait(&sem):以原子操作的方式將信號量的值減去1

int sem_post(&sem):以原子操作的方式將信號量的值加上1

3）對信號量進(jìn)行清理

int sem_destory(&sem)

通過信號量模擬2個窗口，10個客人進(jìn)行服務(wù)的過程

樣例：

#include

#include

#include

#include

#include

#include

#include

#include

using namespace std;

int num=10;

sem_t sem;

void *get_service(void *cid)

{

int id= ((int )cid);

if(sem_wait(&sem)==0)

{

sleep(5);

cout<<"customer "<

cout<<"customer "<

sem_post(&sem);

}

return 0;

}

int main()

{

sem_init(&sem,0,2);

pthread_t customer[num];

int flag;

for(int i=0;i

{

int id=i;

flag=pthread_create(&customer[i],NULL,get_service,&id);

if(flag)

{

cout<<"pthread create error"<

return flag;

}else

{

cout<<"customer "<

}

sleep(1);

}

//wait all thread done

for(int j=0;j

{

pthread_join(customer[j],NULL);

}

sem_destroy(&sem);

return 0;

}

分析：信號量的值代表空閑的服務(wù)窗口，每個窗口一次只能服務(wù)一個人，有空閑窗口，開始服務(wù)前，信號量-1，服務(wù)完成后信號量+1